RocketMQ学习(1) 快速入门

mq的一些前置知识和概念知识可以看这篇文章——SpringCloud入门(3) RabbitMQ,比如常见mq的对比等等,这篇文章不再赘述。

目录

  • RocketMQ概念、安装与配置
    • docker配置
  • RocketMQ快速入门
    • **同步消息
      • 消费模式
    • **异步消息
    • *单向消息
    • **延迟消息
    • *顺序消息
    • 批量消息
    • 事务消息
    • *发送带标签的消息,消息过滤
    • **RocketMQ中消息的Key

RocketMQ概念、安装与配置

Producer:消息的发送者,生产者;举例:发件人
Consumer:消息接收者,消费者;举例:收件人
Broker:暂存和传输消息的通道;举例:快递
NameServer:管理Broker;举例:各个快递公司的管理机构 相当于broker的注册中心,保留了broker的信息
Queue:队列,消息存放的位置,一个Broker中可以有多个队列
Topic:主题,消息的分类
ProducerGroup:生产者组
ConsumerGroup:消费者组,多个消费者组可以同时消费一个主题的消息,同一个组内的消费者订阅关系必须一致。一份消息会传递给每个组,可以被多个消费者组消费,至于组内是广播还是定向则可以自己配置。

消息发送的流程是,Producer询问NameServer,NameServer分配一个broker 然后Consumer也要询问NameServer,得到一个具体的broker,然后消费消息

在这里插入图片描述

在这里插入图片描述



了解了mq的基本概念和角色以后,我们开始安装rocketmq,建议在linux上.下载地址

注意选择版本,这里我们选择4.9.2的版本,后面使用alibaba时对应。source是源码版本,也可以下载下来学习。

在这里插入图片描述

然后上传服务器在root目录下创建文件夹mkdir rocketmq,将下载后的压缩包上传到阿里云服务器或者虚拟机中去.解压unzip rocketmq-all-4.9.2-bin-release.zip解压,如果你的服务器没有unzip命令,则下载安装一个yum install unzip

Benchmark:包含一些性能测试的脚本;Bin:可执行文件目录;Conf:配置文件目录;Lib:第三方依赖LICENSE:授权信息;NOTICE:版本公告;

在这里插入图片描述

然后配置环境变量vim /etc/profile,在文件末尾添加export NAMESRV_ADDR=阿里云公网IP:9876,如果你是云服务器比如阿里云你要写公网地址。比如我这里用的虚拟机,可以填localhost:9876 修改完之后source /etc/profile刷新一下

然后进入bin目录下,观察mqnamesrv启动项运行文件可知,启动项最终真正运行的是runserver.sh文件:

在这里插入图片描述

再进入到runserver.sh,我们发现如果使用的是JDK8,配置的JVM要4个G的内存,最大内存可到8个G,这样我们用于学习的虚拟机或者云服务器够呛能顶住。

在这里插入图片描述

所以我们需要修改一下配置,也用不到那么多的内存.修改runserver.sh文件,将71行和76行的Xms和Xmx等改小一点。vim runserver.sh,:set nu可以让vim看到行号,修改后:wq保存退出。没学过vim的同学建议学一下,后端必备的知识。

在这里插入图片描述


同理,观察mqbroker文件也是一样,启动的是runbroker.sh,我们修改67行,修改内存:

在这里插入图片描述

最后还要修改broker的配置文件,进入conf目录下,修改broker.conf文件

brokerClusterName = DefaultCluster
brokerName = broker-a
brokerId = 0
deleteWhen = 04
fileReservedTime = 48
brokerRole = ASYNC_MASTER
flushDiskType = ASYNC_FLUSH
#添加如下字段
namesrvAddr=localhost:9876
autoCreateTopicEnable=true
brokerIP1=阿里云公网IP

添加参数解释
namesrvAddr:nameSrv地址 可以写localhost nameSrv和broker可以在一个服务器也可以不在
autoCreateTopicEnable:发送消息时如果没有这个Topic则自动创建主题,不然需要手动创建出来,很舒服
brokerIP1:broker也需要一个公网ip,如果不指定,那么是云服务器的内网地址,我们在本地无法连接使用。注意这里不能写成localhost!因为broker把自己的地址登记到name server上,等到发送方去name server找的时候如果写localhost,那肯定连不上。



完成配置之后我们就可以启动了,首先在安装目录下创建一个logs文件夹,mkdir logs,用于存放日志

在这里插入图片描述

先启动nameserver,如果直接./mqnamesrv,这个是前台运行,但我们想后台运行,在java里我们是怎样后台运行呢?我们是nohup java -jar xxx.jar &,我们这里也差不多,在bin目录下运行nohup sh mqnamesrv > ../logs/namesrv.log &>后面指定日志输出的位置

运行完之后怎么看呢?java里是怎么看?ps -ef |grep java,但是这样太麻烦了,可以直接使用jps命令查看java进程,只要安装了jdk就有这个命令,就可以看到本地虚拟机唯一ID(pid即进程号)和Java虚拟机进程的执行主类。这样看的可能不详细,我们使用jps -l,可以直接看到它所有的包名:

在这里插入图片描述

一定要先启动nameserver再启动broker。启动broker,也是一样的,nohup sh mqbroker -c ../conf/broker.conf > ../logs/broker.log &-c后面跟配置文件,>后面指定日志输出的位置。



启动之后我们还想要一个可视化窗口来看,Rocketmq 控制台可以可视化MQ的消息发送,旧版本源码是在rocketmq-external里的rocketmq-console,新版本已经单独拆分成,dashboard,下载地址。

下载后解压出来,在跟目录下执行mvn clean package -Dmaven.test.skip=true,在target目录下得到rocketmq-dashboard-1.0.0.jar,将jar包上传到服务器上去,然后运行nohup java -jar rocketmq-dashboard-1.0.0.jar --server.port=8001 --rocketmq.config.namesrvAddr=127.0.0.1:9876 > ./rocketmq-4.9.2/logs/dashboard.log &

其中--server.port我们指定运行的端口,否则默认是8080
--rocketmq.config.namesrvAddr=127.0.0.1:9876 指定namesrv地址
然后访问IP(我这里是用虚拟机所以是虚拟机地址):8001

在这里插入图片描述

注意如果启动失败的话,要学会去看日志,别忘记我们在启动nameserver、broker还有dashboard时都指定了生成的日志路径,日志里的记录大部分情况下足够dubug了。

Tips:还有一些可能会用到的命令
jps -l查看到进程号之后 如果想重启服务 可以使用kill 进程号来请求终止进程
因为nameserver、broker还有dashboard都很占内存,所以通过free -mh或命令查看一下内存使用情况,如果total-used,即剩余内存大于20%就是安全的。



docker配置

上面的安装流程只是为了入门,真正实际开发大部分还是docker部署,所以这里在贴一下docker部署的过程,docker学习可以参考这篇博客。

下载RockerMQ需要的镜像:docker pull rocketmqinc/rocketmqdocker pull styletang/rocketmq-console-ng

创建NameServer数据存储路径mkdir -p /home/rocketmq/data/namesrv/logs /home/rocketmq/data/namesrv/store

启动NameServer容器docker run -d --name rmqnamesrv -p 9876:9876 -v /home/rocketmq/data/namesrv/logs:/root/logs -v /home/rocketmq/data/namesrv/store:/root/store -e "MAX_POSSIBLE_HEAP=100000000" rocketmqinc/rocketmq sh mqnamesrv

创建Broker数据存储路径mkdir -p /home/rocketmq/data/broker/logs /home/rocketmq/data/broker/store

创建conf配置文件目录mkdir /home/rocketmq/conf

在配置文件目录下创建broker.conf配置文件

# 所属集群名称,如果节点较多可以配置多个
brokerClusterName = DefaultCluster
#broker名称,master和slave使用相同的名称,表明他们的主从关系
brokerName = broker-a
#0表示Master,大于0表示不同的slave
brokerId = 0
#表示几点做消息删除动作,默认是凌晨4点
deleteWhen = 04
#在磁盘上保留消息的时长,单位是小时
fileReservedTime = 48
#有三个值:SYNC_MASTER,ASYNC_MASTER,SLAVE;同步和异步表示Master和Slave之间同步数据的机制;
brokerRole = ASYNC_MASTER
#刷盘策略,取值为:ASYNC_FLUSH,SYNC_FLUSH表示同步刷盘和异步刷盘;SYNC_FLUSH消息写入磁盘后才返回成功状态,ASYNC_FLUSH不需要;
flushDiskType = ASYNC_FLUSH
# 设置broker节点所在服务器的ip地址
brokerIP1 = 你服务器外网ip

启动Broker容器docker run -d --name rmqbroker --link rmqnamesrv:namesrv -p 10911:10911 -p 10909:10909 -v /home/rocketmq/data/broker/logs:/root/logs -v /home/rocketmq/data/broker/store:/root/store -v /home/rocketmq/conf/broker.conf:/opt/rocketmq-4.4.0/conf/broker.conf --privileged=true -e "NAMESRV_ADDR=namesrv:9876" -e "MAX_POSSIBLE_HEAP=200000000" rocketmqinc/rocketmq sh mqbroker -c /opt/rocketmq-4.4.0/conf/broker.conf

启动控制台docker run -d --name rmqadmin -e "JAVA_OPTS=-Drocketmq.namesrv.addr=你的外网地址:9876 \ -Dcom.rocketmq.sendMessageWithVIPChannel=false \ -Duser.timezone='Asia/Shanghai'" -v /etc/localtime:/etc/localtime -p 9999:8080 styletang/rocketmq-console-ng

正常启动后的docker ps:

在这里插入图片描述

访问控制台,你的服务器外网ip:9999

在这里插入图片描述




RocketMQ快速入门

RocketMQ提供了发送多种发送消息的模式,例如同步消息,异步消息,顺序消息,延迟消息,事务消息等,我们一一学习。但是我们先搞清楚消息发送和监听的流程,然后我们在开始敲代码

消息生产者:

1.创建消息生产者producer,并制定生产者组名
2.指定Nameserver地址
3.启动producer
4.创建消息对象,指定主题Topic、Tag和消息体等
5.发送消息
6.关闭生产者producer

消息消费者:

1.创建消费者consumer,制定消费者组名
2.指定Nameserver地址
3.创建监听订阅主题Topic和Tag等
4.处理消息
5.启动消费者consumer


**同步消息

发送同步消息,发送过后会有一个返回值,也就是mq服务器接收到消息后返回的一个确认,生产者在发送后会等待mq主机的返回确认,这种方式非常安全,但是性能上并没有这么高,而且在mq集群中,也是要等到所有的从机都复制了消息以后才会返回,所以针对重要的消息可以选择这种方式,因为几乎任何的mq都会存在消息丢失的风险。

在这里插入图片描述

搭建一个Rocketmq-demo,引入依赖:

<!-- 原生的api   -->
<dependency>
    <groupId>org.apache.rocketmq</groupId>
    <artifactId>rocketmq-client</artifactId>
    <version>4.9.2</version>
    <!--docker的用下面这个版本-->
	<version>4.4.0</version>
</dependency>
<dependency>
    <groupId>junit</groupId>
    <artifactId>junit</artifactId>
    <version>4.12</version>
</dependency>
<dependency>
    <groupId>org.projectlombok</groupId>
    <artifactId>lombok</artifactId>
    <optional>true</optional>
</dependency>

编写生产者

@Test
public void simpleProducer() throws Exception {
    // 创建一个生产者 使用建默认的生产者  (制定一个组名)
    DefaultMQProducer producer = new DefaultMQProducer("test-producer-group");
    // 连接namesrv 设置nameServer地址
    producer.setNamesrvAddr(MqConstant.NAME_SRV_ADDR); // 字符常量 ip:9876
    // 启动
    producer.start();
    // 创建一个消息
    for (int i = 0; i < 10; i++) {
        // 创建消息 第一个参数:主题的名字 第二个参数:消息内容
        Message message = new Message("testTopic", "我是一个简单的消息".getBytes());
        // 发送消息 有返回值 可以打印一下
        SendResult sendResult = producer.send(message);
        System.out.println(sendResult.getSendStatus());
    }
    // 关闭生产者
    producer.shutdown();
}

编写消费者

 消费者
@Test
public void simpleConsumer() throws Exception {
    // 创建一个消费者 默认消费者组  (制定一个组名)
    DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("test-consumer-group");
    // 连接namesrv 设置nameServer地址
    consumer.setNamesrvAddr(MqConstant.NAME_SRV_ADDR); // 字符常量 ip:9876
    // 订阅一个主题  写入主题名和消息表达式 *标识订阅这个主题中所有的消息  后期再介绍消息过滤
    consumer.subscribe("testTopic", "*");
    // 设置一个监听器 (一直监听的, 异步回调方式,消费线程和主线程不是一个线程)
    // MessageListenerConcurrently 是多线程消费,默认20个线程,可以通过 consumer.setConsumeThreadMax() 来设置最大消费线程数
    consumer.registerMessageListener(new MessageListenerConcurrently() {
        @Override
        public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
            // 这个就是消费的方法 (业务处理)
            System.out.println("我是消费者");
            // 这里的消息虽然是个List 但默认只有一个消息 后面有配置批量消息这样的概念
            System.out.println(msgs.get(0).toString());
            System.out.println("消息内容:" + new String(msgs.get(0).getBody()));
            System.out.println("消费上下文:" + context);
            // 返回消费的状态 如果是CONSUME_SUCCESS 则成功,若为RECONSUME_LATER则该条消息会被重回队列,重新被投递
            // 重试的时间为messageDelayLevel = "1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h
            // 也就是第一次1s 第二次5s 第三次10s .... 如果重试了16次 那么这个消息就会被终止发送给消费者
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        }
    });
    // 启动 这个start一定要写在registerMessageListener下面
    consumer.start();
    // 主线程不能退出,否则消费者线程也会随之退出 挂起当前 JVM,保持主线程存活,以便消费者能持续运行
    System.in.read(); // 让程序在该行停下来,等待用户输入
}

注意MessageListenerConcurrently 是多线程消费,默认20个线程,可以通过 consumer.setConsumeThreadMax() 来设置最大消费线程数。所以当有消息同时进来了,消息的处理是并行的。

测试

发送后消息,输出如下,rocketmq控制面板的主题多了一个我们刚刚发送的,我们再来详细看一下主题里的这些按钮的一些功能。

在这里插入图片描述

在这里插入图片描述

查看其状态可知,一个broker可以有不同的topic,我们这里只配了一个broker,一个topic默认有4个队列,采用默认的负载均衡算法往里面去放消息,看如下图可知,消息分配是很均匀的。

这里补充一点知识,如果是广播模式,那好说,一个组内所有的消费者都会拿到每个队列里的消息,如果是负载均衡模式,假设有c1、c2两个消费者,那么每个队列是要固定联系好消费者的,即比如队列1、2的消息只会给c1,2、3的消息只会给c2,比如4个队列都指定了消费者,组内再有一个消费者,那它永远没有消息。
所以最好队列数量>=组内消费者数量

在这里插入图片描述

然后去看它的路由,主要是broker的地址,先通过nameserver找到broker,然后通过broker连接发送的消息。这里还有读队列写队列还有权限,先不管后面再介绍。

在这里插入图片描述

点击consumer管理可以发现现在还没有订阅者,

在这里插入图片描述

topic配置可以修改一些配置:

在这里插入图片描述

也可以通过面板发送消息,但一般不这样做。

在这里插入图片描述

还有重置消费位点,跳过堆积,删除这三个操作,我们后面介绍。



测试消费代码:
中间就是消息的内容,里面有很多东西,broker的名字,队列号,队列偏移量,还有一些状态、时间,消息内容在body里以字节数组存放着

在这里插入图片描述

此时ui控制台的消费者管理就能看到消费者了,消费者都是一个人,这里的代理者位点、消费者位点、差值是什么意思呢?
mq是代理者,我们的程序就是消费者,说白了就是4个队列中的消息,总共接收、已消费、未消费的消息数量。每消息成功消息一次,消费者位点就移动增加一位,队列每接收到一条消息,代理者位点就增加一位。

在这里插入图片描述

消费模式

MQ的消费模式可以大致分为两种,一种是推Push,一种是拉Pull。我们刚刚写的代码都是Push–>DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("test-consumer-group"); 现在官方不推荐使用Pull了–>DefaultMQPullConsumer consumer = new DefaultMQPullConsumer("test-consumer-group");

官方推荐不再使用Pull API的原因
复杂性和易用性: Pull模式需要用户手动管理拉取逻辑,包括拉取频率、拉取数量、异常处理、消息重试等。这增加了实现的复杂性,不容易使用和维护。相反,Push模式通过封装这些复杂性,提供了更高层次的API,简化了开发工作。
性能和资源管理: 在Push模式中,RocketMQ客户端能够更好地管理和优化拉取逻辑,包括流控和负载均衡,从而提高整体性能和资源利用率。手动实现这些优化在Pull模式中是比较困难的。
一致性和可靠性: Push模式内置了很多一致性和可靠性的保障机制,例如消费确认、重试机制等。这些在Pull模式中需要用户自己实现,容易出错。

  • Push是服务端【MQ】主动推送消息给客户端,优点是及时性较好,但如果客户端没有做好流控,一旦服务端推送大量消息到客户端时,就会导致客户端消息堆积甚至崩溃。

  • Pull是客户端需要主动到服务端取数据,优点是客户端可以依据自己的消费能力进行消费,但拉取的频率也需要用户自己控制,拉取频繁容易造成服务端和客户端的压力,拉取间隔长又容易造成消费不及时。

其实Push模式也是基于Pull模式的,只是客户端内部封装了api,通过长轮询方式实现的,使得消费者不需要手动拉取,而是通过回调或者监听的方式获取消息。因为任何mq中间件都是Pull模式,都需要你主动去拉,它不会主动推给你,mq的qps很高,一有消息全部都主动推给你服务端肯定受不了的。

总结:一般场景下,上游消息生产量小或者均速的时候,可以选择Push模式。在特殊场景下,例如电商大促,抢优惠券等场景可以选择Pull模式。



**异步消息

异步消息通常用在对响应时间敏感的业务场景,即发送端不能容忍长时间地等待Broker的响应。发送完以后会有一个异步消息通知

编写生产者

@Test
public void asyncProducer() throws Exception {
    // 创建默认的生产者
    DefaultMQProducer producer = new DefaultMQProducer("async-producer-group");
    producer.setNamesrvAddr(MqConstant.NAME_SRV_ADDR); // 设置nameServer地址
    producer.start(); // 启动实例
    Message message = new Message("asyncTopic", "我是一个异步消息".getBytes());
    producer.send(message, new SendCallback() {
        @Override
        public void onSuccess(SendResult sendResult) {
            System.out.println("发送成功");
        }

        @Override
        public void onException(Throwable e) {
            System.err.println("发送失败:" + e.getMessage());
        }
    });
    System.out.println("我先执行");
    System.in.read(); // 挂起jvm 因为回调是异步的不然测试不出来 键入回车键就解除挂起
    producer.shutdown(); // 关闭实例
}

消费者的代码可与同步消息使用同一个。



*单向消息

这种方式主要用在不关心发送结果的场景,这种方式吞吐量很大,但是存在消息丢失的风险,例如日志信息的发送

生产者代码

@Test
public void onewayProducer() throws Exception {
    // 创建默认的生产者
    DefaultMQProducer producer = new DefaultMQProducer("oneway-producer-group");
    producer.setNamesrvAddr(MqConstant.NAME_SRV_ADDR); // 创建默认的生产者
    producer.start(); // 启动实例
    Message message = new Message("onewayTopic", "日志xxx".getBytes());
    producer.sendOneway(message); // 发送单向消息
    System.out.println("成功");
    producer.shutdown(); // 关闭实例
}

消费者代码同上,测试的时候注意主题topic更换

拓充一下日志采集的思路
开发环境,直接输出在控制台即可
生产环境下,可以输出在文件,但是查看起来就很麻烦了,那么可以输出到mysql,但是存储的内容有限,后期我们还会学到ES,存储的数据量级就很大了、
再开扩一下思路,每个业务操作都要记录日志,虽然麻烦但是要有,那么每个操作都有一个插入日志到mysql或者es的操作不就很费时了吗?那么可以专门为日志操作搭建一个服务器log-service,业务操作是生产者,再有一个消费者来记录日志,mq是异步的,这样性能也好很多。



**延迟消息

消息放入mq后,过一段时间,才会被监听到,然后消费
比如下订单业务,提交了一个订单就可以发送一个延时消息,30min后去检查这个订单的状态,如果还是未付款就取消订单释放库存。

生产者代码

@Test
public void msProducer() throws Exception {
    // 创建默认的生产者
    DefaultMQProducer producer = new DefaultMQProducer("ms-producer-group");
    producer.setNamesrvAddr(MqConstant.NAME_SRV_ADDR); // 设置nameServer地址
    producer.start(); // 启动实例
    Message message = new Message("orderMsTopic", "订单号,座位号".getBytes());
    // 给消息设置一个延迟时间
    // messageDelayLevel = "1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h"
    message.setDelayTimeLevel(3); // 10s延迟
    // 发延迟消息 这里没有获取返回值 所以是单向的
    producer.send(message);
    System.out.println("发送时间" + new Date()); // 打印时间
    producer.shutdown(); // 关闭实例
}

消费者代码同上,可以自己改一下测试一下,在消费方法里打印一下当前时间,测试发现其实不是很精确的10s,与官方的时间还是有误差的,误差还是很大,我自己的环境测试26s才接收到,因为broker默认是8个G的内存,我们给修改成256m了,性能不足。第一次测试比较慢,第二次就比较准了,因为第一次还需要加载一些东西,冷加载嘛。

发送时间Mon May 27 16:36:24 CST 2024

收到消息时间:Mon May 27 16:36:50 CST 2024
订单号,座位号

# 第二次测试

发送时间Mon May 27 16:43:56 CST 2024

收到消息时间:Mon May 27 16:44:06 CST 2024
订单号,座位号


*顺序消息

消息有序指的是可以按照消息的发送顺序来消费(FIFO)。RocketMQ可以严格的保证消息有序,可以分为:分区有序或者全局有序。
可能大家会有疑问,mq不就是FIFO吗?
rocketMq的broker的机制,导致了rocketMq会有这个问题
因为一个broker中对应了四个queue

在这里插入图片描述

在默认的情况下消息发送会采取Round Robin轮询方式把消息发送到不同的queue(分区队列);而消费消息的时候以多线程的方式从多个queue上拉取消息,这种情况发送和消费是不能保证顺序,所以肯定要修改为以单线程的方式去消费。

如果控制发送的顺序消息只依次发送到同一个queue中,消费的时候只从这个queue上依次拉取,则就保证了顺序。当发送和消费参与的queue只有一个,则是全局有序;但也没必要全局有序,我们只需要确保局部有序就行了,如果多个queue参与,各个queue内的消息是相对有序的,则为分区有序,即相对每个queue,消息都是有序的。

下面用订单进行分区有序的示例。一个订单的顺序流程是:下订单、发短信通知、物流、签收,必须依次执行。订单顺序号相同的消息会被先后发送到同一个队列中,消费时,同一个顺序获取到的肯定是同一个队列。

模拟一个订单的发送流程,创建两个订单,发送的消息分别是
订单号111 消息流程 下订单->物流->签收
订单号112 消息流程 下订单->物流->拒收

先创建一个订单对象

@Data
@AllArgsConstructor
@NoArgsConstructor
public class MsgModel {

    private String orderSn;
    private Integer userId;
    private String desc; // 下单 短信 物流

    // xxx
}

生产者代码


其中发送消息有个很重要的api,用于消息的选择,我们ctrl+P可以查看:

在这里插入图片描述

生产者代码

@Test
public void testOrderlyProducer() throws Exception {
    // 创建默认的生产者
    DefaultMQProducer producer = new DefaultMQProducer("orderly-producer-group");
    producer.setNamesrvAddr(MqConstant.NAME_SRV_ADDR); // 设置nameServer地址
    producer.start(); // 启动实例
    List<Order> orderList = Arrays.asList(
            new Order(1, 111, 59D, new Date(), "下订单"),
            new Order(2, 111, 59D, new Date(), "物流"),
            new Order(3, 111, 59D, new Date(), "签收"),
            new Order(4, 112, 89D, new Date(), "下订单"),
            new Order(5, 112, 89D, new Date(), "物流"),
            new Order(6, 112, 89D, new Date(), "拒收")
    );
    // 循环集合开始发送 发送顺序消息发送时要确保有序 并且要发到同一个队列下面去
    orderList.forEach(order -> {
        Message message = new Message("orderlyTopic", order.toString().getBytes());
        try {
            // 发 相同的订单号去相同的队列
            producer.send(message, new MessageQueueSelector() {
                @Override
                public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) { // 在这里 选择队列
                    // 这里重新函数的mqs就是所有的队列(默认是4个) msg就是这条消息 arg是send函数的第三个参数传入进来的
                    int queueNumber = mqs.size(); // 当前主题有多少个队列
                    // 这个arg就是后面传入的 order.getOrderNumber()
                    Integer i = (Integer) arg;
                    int index = i % queueNumber; // 用这个值去%队列的个数得到一个队列
                    // 返回选择的这个队列即可 ,那么相同的订单号 就会被放在相同的队列里 实现FIFO了
                    return mqs.get(index); // 选择发送到第几个队列
                }
            }, order.getOrderNumber()); // msgModel.getOrderSn()是send函数的第三个参数 这个值传给了select函数的arg
        } catch (Exception e) {
            System.out.println("发送异常");
        }
    });
    producer.shutdown(); // 关闭实例
    System.out.println("发送完成");
}

发送完可以看到消息被均匀的放在了两个队列里:

在这里插入图片描述

消费者代码,消费者也有讲究,也是刚刚提到的,不能再用并发模式了,而是采用单线程去消费,或者采用多线程,将最大线程数设置为1也行。
注意这里的单线程不是全局只有一个,是针对每个队列单独使用一个线程进行顺序消费,而不是整个消费者实例只有一个线程,我们可以打印一下线程id来看一下。

@Test
public void orderlyConsumer() throws Exception {
    // 创建默认消费者组
    DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("orderly-consumer-group");
    consumer.setNamesrvAddr(MqConstant.NAME_SRV_ADDR); // 设置nameServer地址
    // 订阅一个主题来消费 *表示没有过滤参数 表示这个主题的任何消息
    consumer.subscribe("orderlyTopic", "*");
    // MessageListenerConcurrently 并发模式 多线程的  重试16次
    // MessageListenerOrderly 顺序模式 单线程的 无限重试Integer.Max_Value
    consumer.registerMessageListener(new MessageListenerOrderly() {
        @Override
        public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext context) {
            System.out.println("线程id:" + Thread.currentThread().getId());
            System.out.println(new String(msgs.get(0).getBody()));
            return ConsumeOrderlyStatus.SUCCESS;
        }
    });
    consumer.start();
    System.in.read();
}

我的这个运行有点太理想了,多尝试几次可以发现,有可能出现队列之间交叉消费,但是不管怎样,队内的顺序是相对有序的。
在这里插入图片描述

针对ConsumeOrderlyStatus,它跟ConsumeConcurrentlyStatus还是有一些区别,我们看一下:

在这里插入图片描述

然后消费者代码里这个SUSPEND_CURRENT_QUEUE_A_MOMENT意思就是消费失败了,队列里有顺序消息A B C,A消费失败了,不会去消费B和C,而是挂起当前的A消息,等一会再去消费,如果A一直报错,则会一直重试,直到A成功才去消费B,

关于重试时间和次数,可以去查看一下DefaultMQPushConsumer的源码,默认重试间隔为1s,重试次数无限制。可以自己设置每次重试的时间间隔(单位:毫秒) consumer.setSuspendCurrentQueueTimeMillis(2000);// 2秒

设置最大重试次数 consumer.setMaxReconsumeTimes(5); // 最大重试5次



批量消息

Rocketmq可以一次性发送一组消息,那么这一组消息会被当做一个消息消费

这种方式发送的消息,可以测试一下,在前端仪表板看看主题状态,发送的这些打包的消息都是放在一个队列里的,但是消息还是有三个的,等于说只有发是打包的,可以测试一下消费端,输出一下接收的消息的长度,还是只有一个。

生产者代码

@Test
public void testBatchProducer() throws Exception {
    // 创建默认的生产者
    DefaultMQProducer producer = new DefaultMQProducer("batch-producer-group");
    // 设置nameServer地址
    producer.setNamesrvAddr(MqConstant.NAME_SRV_ADDR);
    // 启动实例
    producer.start();
    List<Message> msgs = Arrays.asList(
            new Message("batchTopic", "我是一组消息的A消息".getBytes()),
            new Message("batchTopic", "我是一组消息的B消息".getBytes()),
            new Message("batchTopic", "我是一组消息的C消息".getBytes())
    );
    SendResult send = producer.send(msgs);
    System.out.println(send);
    // 关闭实例
    producer.shutdown();
}

消费者代码

@Test
public void msConsumer() throws Exception {
    // 创建默认消费者组
    DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("batch-consumer-group");
    // 设置nameServer地址
    consumer.setNamesrvAddr(MqConstant.NAME_SRV_ADDR);
    // 订阅一个主题来消费和订阅消息表达式,默认是*
    consumer.subscribe("batchTopic", "*");
    // 注册一个消费监听 MessageListenerConcurrently是并发消费
    // 默认是20个线程一起消费,可以通过 consumer.setConsumeThreadMax() 来设置
    consumer.registerMessageListener(new MessageListenerConcurrently() {
        @Override
        public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
            // 这里执行消费的代码 默认是多线程消费
            System.out.println(Thread.currentThread().getName() + "----" + new String(msgs.get(0).getBody()));
            System.out.println("收到消息了" + new Date());
            System.out.println(msgs.size());
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        }
    });
    consumer.start();
    System.in.read();
}


事务消息

这个可以先不学了,比较鸡肋,基本不用,后面我们会专门学分布式事务seata,它是专门解决分布式事务问题的。这里也附上代码,有兴趣可以学一下。

事务消息的发送流程。它可以被认为是一个两阶段的提交消息实现,以确保分布式系统的最终一致性。事务性消息确保本地事务的执行和消息的发送可以原子地执行。

在这里插入图片描述

在这里插入图片描述

上图说明了事务消息的大致方案,其中分为两个流程:正常事务消息的发送及提交、事务消息的补偿流程。


事务消息发送及提交

  1. 发送消息(half消息)。
  2. 服务端响应消息写入结果。
  3. 根据发送结果执行本地事务(如果写入失败,此时half消息对业务不可见,本地逻辑不执行)。
  4. 根据本地事务状态执行Commit或Rollback(Commit操作生成消息索引,消息对消费者可见)

事务补偿

  1. 对没有Commit/Rollback的事务消息(pending状态的消息),从服务端发起一次“回查”
  2. Producer收到回查消息,检查回查消息对应的本地事务的状态
  3. 根据本地事务状态,重新Commit或者Rollback
    其中,补偿阶段用于解决消息UNKNOW或者Rollback发生超时或者失败的情况。

事务消息状态

事务消息共有三种状态,提交状态、回滚状态、中间状态:
 TransactionStatus.CommitTransaction: 提交事务,它允许消费者消费此消息。
 TransactionStatus.RollbackTransaction: 回滚事务,它代表该消息将被删除,不允许被消费。
 TransactionStatus.Unknown: 中间状态,它代表需要检查消息队列来确定状态。

事务消息生产者代码

/**
 * TransactionalMessageCheckService的检测频率默认1分钟,可通过在broker.conf文件中设置transactionCheckInterval的值来改变默认值,单位为毫秒。
 * 从broker配置文件中获取transactionTimeOut参数值。
 * 从broker配置文件中获取transactionCheckMax参数值,表示事务的最大检测次数,如果超过检测次数,消息会默认为丢弃,即回滚消息。
 *
 * @throws Exception
 */
@Test
public void testTransactionProducer() throws Exception {
    // 创建一个事务消息生产者
    TransactionMQProducer producer = new TransactionMQProducer("test-group");
    producer.setNamesrvAddr("localhost:9876");
    // 设置事务消息监听器
    producer.setTransactionListener(new TransactionListener() {
        // 这个是执行本地业务方法
        @Override
        public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {
            System.out.println(new Date());
            System.out.println(new String(msg.getBody()));
            // 这个可以使用try catch对业务代码进行性包裹
            // COMMIT_MESSAGE 表示允许消费者消费该消息
            // ROLLBACK_MESSAGE 表示该消息将被删除,不允许消费
            // UNKNOW表示需要MQ回查才能确定状态 那么过一会 代码会走下面的checkLocalTransaction(msg)方法
            return LocalTransactionState.UNKNOW;
        }

        // 这里是回查方法 回查不是再次执行业务操作,而是确认上面的操作是否有结果
        // 默认是1min回查 默认回查15次 超过次数则丢弃打印日志 可以通过参数设置
        // transactionTimeOut 超时时间
        // transactionCheckMax 最大回查次数
        // transactionCheckInterval 回查间隔时间单位毫秒
        // 触发条件
        // 1.当上面执行本地事务返回结果UNKNOW时,或者下面的回查方法也返回UNKNOW时 会触发回查
        // 2.当上面操作超过20s没有做出一个结果,也就是超时或者卡主了,也会进行回查
        @Override
        public LocalTransactionState checkLocalTransaction(MessageExt msg) {
            System.err.println(new Date());
            System.err.println(new String(msg.getBody()));
            // 这里
            return LocalTransactionState.UNKNOW;
        }
    });
    producer.start();
    Message message = new Message("TopicTest2", "我是一个事务消息".getBytes());
    // 发送消息
    producer.sendMessageInTransaction(message, null);
    System.out.println(new Date());
    System.in.read();
}

事务消费者代码

@Test
public void testTransactionConsumer() throws Exception {
    // 创建默认消费者组
    DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("consumer-group");
    // 设置nameServer地址
    consumer.setNamesrvAddr("localhost:9876");
    // 订阅一个主题来消费 *表示没有过滤参数 表示这个主题的任何消息
    consumer.subscribe("TopicTest2", "*");
    // 注册一个消费监听 MessageListenerConcurrently是并发消费
    // 默认是20个线程一起消费,可以参看 consumer.setConsumeThreadMax()
    consumer.registerMessageListener(new MessageListenerConcurrently() {
        @Override
        public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
                                                        ConsumeConcurrentlyContext context) {
            // 这里执行消费的代码 默认是多线程消费
            System.out.println(Thread.currentThread().getName() + "----" + new String(msgs.get(0).getBody()));
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        }
    });
    consumer.start();
    System.in.read();
}

测试结果

在这里插入图片描述



*发送带标签的消息,消息过滤

Rocketmq提供消息过滤功能,通过tag或者key进行区分
我们往一个主题里面发送消息的时候,根据业务逻辑,可能需要区分,比如带有tagA标签的被A消费,带有tagB标签的被B消费,还有在事务监听的类里面,只要是事务消息都要走同一个监听,我们也需要通过过滤才区别对待

什么时候该用 Topic,什么时候该用 Tag?
总结:不同的业务应该使用不同的Topic如果是相同的业务里面有不同表的表现形式,那么我们要使用tag进行区分

可以从以下几个方面进行判断:
1.消息类型是否一致:如普通消息、事务消息、定时(延时)消息、顺序消息,不同的消息类型使用不同的 Topic,无法通过 Tag 进行区分。
2.业务是否相关联:没有直接关联的消息,如淘宝交易消息,京东物流消息使用不同的 Topic 进行区分;而同样是天猫交易消息,电器类订单、女装类订单、化妆品类订单的消息可以用 Tag 进行区分。
3.消息优先级是否一致:如同样是物流消息,盒马必须小时内送达,天猫超市 24 小时内送达,淘宝物流则相对会慢一些,不同优先级的消息用不同的 Topic 进行区分。
4.消息量级是否相当:有些业务消息虽然量小但是实时性要求高,如果跟某些万亿量级的消息使用同一个 Topic,则有可能会因为过长的等待时间而“饿死”,此时需要将不同量级的消息进行拆分,使用不同的 Topic。
总的来说,针对消息分类,您可以选择创建多个 Topic,或者在同一个 Topic 下创建多个 Tag。但通常情况下,不同的 Topic 之间的消息没有必然的联系,而 Tag 则用来区分同一个 Topic 下相互关联的消息,例如全集和子集的关系、流程先后的关系。

在这里插入图片描述

在这里插入图片描述

生产者代码

@Test
public void tagProducer() throws Exception {
    // 创建默认的生产者
    DefaultMQProducer producer = new DefaultMQProducer("tag-producer-group");
    producer.setNamesrvAddr(MqConstant.NAME_SRV_ADDR); // 设置nameServer地址
    producer.start(); // 启动实例

    Message message = new Message("tagTopic", "vip1", "我是vip1的文章".getBytes());
    Message message2 = new Message("tagTopic", "vip2", "我是vip2的文章".getBytes());
    producer.send(message);
    producer.send(message2);
    System.out.println("发送成功");
    producer.shutdown(); // 关闭实例
}

消费者代码

/**
 * vip1
 *
 * @throws Exception
 */
@Test
public void tagConsumer1() throws Exception {
    DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("tag-consumer-group-a");
    consumer.setNamesrvAddr(MqConstant.NAME_SRV_ADDR);
    consumer.subscribe("tagTopic", "vip1");
    consumer.registerMessageListener(new MessageListenerConcurrently() {
        @Override
        public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
            System.out.println("我是vip1的消费者,我正在消费消息" + new String(msgs.get(0).getBody()));
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        }
    });
    consumer.start();
    System.in.read();
}


/**
 * vip1 || vip2
 *
 * @throws Exception
 */
@Test
public void tagConsumer2() throws Exception {
    DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("tag-consumer-group-b");
    consumer.setNamesrvAddr(MqConstant.NAME_SRV_ADDR);
    consumer.subscribe("tagTopic", "vip1 || vip2");
    consumer.registerMessageListener(new MessageListenerConcurrently() {
        @Override
        public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
            System.out.println("我是vip2的消费者,我正在消费消息" + new String(msgs.get(0).getBody()));
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        }
    });
    consumer.start();
    System.in.read();
}

查看官方文档,可知订阅关系的定义:一个消费者组订阅一个Topic的某一个Tag,这种记录被称之为订阅关系。

注意!只有订阅关系完全一致才能称之为组,比如这里组a的订阅关系是主题tagTopic中的tag:vip1,组b的订阅关系是主题tagTopic中的tag:vip1||vip2,所以它两不能是同一个消费者组。如果订阅关系不一致,会导致消费消息紊乱,甚至消息丢失。

对于消费者组,我们之间提到:消费者组,多个消费者组可以同时消费一个主题的消息,同一个组内的消费者订阅关系必须一致。一份消息会传递给每个组,至于组内是广播还是定向则可以自己配置。

这部分逻辑强烈推荐阅读一下官方文档,其实这类技术还是学习官方文档上手更快。

所以测试这里的代码,vip1消息这两个消费者都能接收到,vip2消息则只有第二个消费者能接收到了。

想到如果消费者2隔了很久再去消费还能接收到这条消息吗,去查了相关资料:
RocketMQ的消息是持久化存储在Broker(消息代理)中的。默认情况下,RocketMQ会将消息存储7天,超过这个时间后消息会被自动删除。当然,这个存储时间是可以配置的,通过broker.conf中的参数fileReservedTime来进行设置。



**RocketMQ中消息的Key

在rocketmq中的消息,默认会有一个messageId当做消息的唯一标识,我们自己也可以给消息携带一个key,用作唯一标识或者业务标识,包括在控制面板查询的时候也可以使用messageId或者key来进行查询

在这里插入图片描述

生产者代码

/**
 * 业务参数 我们自身要确保唯一
 * 为了查阅和去重
 *
 * @throws Exception
 */
@Test
public void keyProducer() throws Exception {
    DefaultMQProducer producer = new DefaultMQProducer("key-producer-group");
    producer.setNamesrvAddr(MqConstant.NAME_SRV_ADDR);
    producer.start();
    String key = UUID.randomUUID().toString(); // 通过UUID来唯一标识
    System.out.println(key);
    Message message = new Message("keyTopic", "vip1", key, "我是vip1的文章".getBytes());
    producer.send(message);
    System.out.println("发送成功");
    producer.shutdown();
}

在这里插入图片描述

消费者代码

@Test
public void keyConsumer() throws Exception {
    DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("key-consumer-group");
    consumer.setNamesrvAddr(MqConstant.NAME_SRV_ADDR);
    consumer.subscribe("keyTopic", "*"); // key不影响订阅关系
    consumer.registerMessageListener(new MessageListenerConcurrently() {
        @Override
        public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
            MessageExt messageExt = msgs.get(0);
            System.out.println("我是vip1的消费者,我正在消费消息" + new String(messageExt.getBody()));
            System.out.println("我们业务的标识:" + messageExt.getKeys()); // key放在消息体里 可以拿到
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        }
    });
    consumer.start();
    System.in.read();
}

messageExt还可以拿到很多东西,可以自己去尝试一下。





本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处:/a/656570.html

如若内容造成侵权/违法违规/事实不符,请联系我们进行投诉反馈qq邮箱809451989@qq.com,一经查实,立即删除!

相关文章

通过提示工程将化学知识整合到大型语言模型中

在当今快速发展的人工智能领域&#xff0c;大型语言模型&#xff08;LLMs&#xff09;正成为科学研究的新兴工具。这些模型以其卓越的语言处理能力和零样本推理而闻名&#xff0c;为解决传统科学问题提供了全新的途径。然而&#xff0c;LLMs在特定科学领域的应用面临挑战&#…

力扣HOT100 - 1143. 最长公共子序列

解题思路&#xff1a; 动态规划 class Solution {public int longestCommonSubsequence(String text1, String text2) {int m text1.length(), n text2.length();int[][] dp new int[m 1][n 1];for (int i 1; i < m; i) {char c1 text1.charAt(i - 1);for (int j 1…

【算法】位运算算法——两整数之和

题解&#xff1a;两整数之和(位运算算法) 目录 1.题目2.位运算算法3.参考代码4.总结 1.题目 题目链接&#xff1a;LINK 2.位运算算法 这个题目难点就在于不能用、- 那什么能够代替加号呢&#xff1f; 既然数的层面不能用号&#xff0c;那二进制的角度去用号即可。 恰好&a…

JavaScript(ES6)入门

ES6 1、介绍 ECMAScript 6&#xff08;简称ES6&#xff09;是于2015年6月正式发布的JavaScript 语言的标准&#xff0c;正式名为ECMAScript 2015&#xff08;ES2015&#xff09;。它的目标是使得JavaScript语言可以用来编写复杂的大型应用程序&#xff0c;成为企业级开发语言。…

AAAI2024 基于扩散模型 多类别 工业异常检测 DiAD

前言 本文分享一个基于扩散模型的多类别异常检测框架&#xff0c;用于检测工业场景的缺陷检测或异常检测。 设计SG语义引导网络&#xff0c;在重建过程中有效保持输入图像的语义信息&#xff0c;解决了LDM在多类别异常检测中的语义信息丢失问题。高效重建&#xff0c;通过在潜…

mysql实战——Mysql8.0高可用之双主+keepalived

一、介绍 利用keepalived实现Mysql数据库的高可用&#xff0c;KeepalivedMysql双主来实现MYSQL-HA&#xff0c;两台Mysql数据库的数据保持完全一致&#xff0c;实现方法是两台Mysql互为主从关系&#xff0c;通过keepalived配置VIP&#xff0c;实现当其中的一台Mysql数据库宕机…

Wpf 使用 Prism 实战开发Day27

首页汇总和数据动态显示 一.创建首页数据汇总数据接口 汇总&#xff1a;待办事项的总数已完成&#xff1a;待办事项里面有多少条完成的待办完成比例&#xff1a;已完成和汇总之间的比例备忘录&#xff1a;显示备忘录的总数待办事项&#xff1a;显示待办事项未完成的集合备忘录&…

Flask+Vue+MySQL天水麻辣烫管理系统设计与实现(附源码 配置 文档)

背景&#xff1a; 同学找到我期望做一个天水麻辣烫的网页&#xff0c;想复用以前做过的课设&#xff0c;结合他的实际需求&#xff0c;让我们来看看这个系统吧~ 项目功能与使用技术概述&#xff1a; 里面嵌入了6个子系统&#xff0c;其中餐饮系统可以进行餐馆信息添加、修改…

【ARFoundation自学03】平面追踪可视化效果美化

对已检测到的平面默认的渲染效果显得有些生硬和突兀&#xff0c;有时我们需要更加友好、美观的的平面虚拟界面&#xff0c;这时就需要对已检测到的平面定制个性化的可视方案。为达到更好的视觉效果&#xff0c;处理的思路如下。 视觉效果前后对比&#xff01; &#xff08;本节…

Android Compose 七:常用组件 Image

1 基本使用 Image(painter painterResource(id R.drawable.ic_wang_lufei), contentDescription "" ) // 图片Spacer(modifier Modifier.height(20.dp))Image(imageVector ImageVector.vectorResource(id R.drawable.ic_android_black_24dp), contentDescript…

Nature 正刊!瑞典于默奥大学研究团队在研究全球河流和溪流的甲烷排放中取得新进展

甲烷(CH4)是一种强有力的温室气体&#xff0c;自工业革命以来&#xff0c;其在大气中的浓度增加了两倍。有证据表明&#xff0c;全球变暖增加了淡水生态系统的 CH4 排放&#xff0c;为全球气候提供了积极的反馈。然而&#xff0c;对于河流和溪流来说&#xff0c;甲烷排放的控制…

618有什么宠物空气净化器推荐?希喂FreAir Lite宠物空气净化器真实体验

一、宠物空气净化器的必要性 掉毛季又来了&#xff0c;猫咪的毛发满天飞&#xff0c;怎么办&#xff1f;我家里的猫咪一到换毛季就掉满地的毛发&#xff0c;尤其喜欢在家里奔跑打闹&#xff0c;结果整个房间都是毛。为了减少家里空气中的浮毛&#xff0c;你都做过哪些努力呢&a…

电脑出现:excel词典(xllex.dll)文件丢失或损坏的错误提示怎么办?有效的将丢失的xllex.dll修复

当遇到 Excel 提示“词典 (xllex.dll) 文件丢失或损坏”的问题时&#xff0c;通常意味着该动态链接库文件&#xff08;Dynamic Link Library&#xff0c;DLL&#xff09;&#xff0c;它与拼写检查功能相关联的&#xff0c;无法被正确找到或者合适地使用。那么有什么办法可以解决…

空间转录组数据的意义

10X空间转录组Visium学习笔记&#xff08;三&#xff09;跑通Visium全流程记录 | 码农家园 (codenong.com) 这两个的区别是&#xff1a;一个是像素的位置信息&#xff0c;一个是阵列的位置信息

第97天:权限提升-Web 权限权限划分源码后台中间件第三方数据库等

前置知识 具体有哪些权限需要我们了解掌握的 后台权限&#xff0c;网站权限&#xff0c;数据库权限&#xff0c;接口权限&#xff0c;系统权限&#xff0c;域控权限等 以上常见权限获取方法简要归类说明 后台权限&#xff1a;SQL 注入,数据库备份泄露&#xff0c;默认或弱口…

设计模型-系统架构师(三)

1、按照《中华人民共和国著作权法》的权利保护期&#xff0c;署名权和修改权 是永久保护的。 发表权、使用权和获取报酬权 保护期限 作者终生和死后的50年。 2、&#xff08;重要&#xff09;面向对象的分析模型主要由 顶层架构图、用例与用例图和&#xff08;&#xff09;构成…

CTF网络安全大赛简单的web抓包题目:HEADache

题目来源于&#xff1a;bugku 题目难度&#xff1a;简单 题目 描  述: > Wanna learn about some types of headache? > Lets dig right into it! 下面是题目源代码&#xff1a; <!DOCTYPE html> <html> <head><meta charset"utf-8"&…

基于C#开发web网页管理系统模板流程-主界面管理员入库和出库功能完善

前言 紧接上篇->基于C#开发web网页管理系统模板流程-主界面管理员录入和编辑功能完善-CSDN博客 本篇将完善主界面的管理员入库和出库功能&#xff0c;同样的&#xff0c;管理员入库和出库的设计套路适用于动态表的录入和编辑 首先还是介绍一下本项目将要实现的功能 &#xf…

对象解构与迭代器的猫腻?

前言 变量的解构赋值是前端开发中经常用到的一个技巧&#xff0c;比如&#xff1a; // 对象解构 const obj { a: 1, b: 2 }; const { a, b } obj; console.log(a, b)数组解构 const arr [1, 2, 3]; const [a, b] arr; console.log(a, b)工作中我们最经常用的就是类似上面…

【Python安全攻防】【网络安全】一、常见被动信息搜集手段

一、IP查询 原理&#xff1a;通过目标URL查询目标的IP地址。 所需库&#xff1a;socket Python代码示例&#xff1a; import socketip socket.gethostbyname(www.163.com) print(ip)上述代码中&#xff0c;使用gethostbyname函数。该函数位于Python内置的socket库中&#xf…