RocketMQ第三节（生产者和消费者）

目录

1：生产者（同步、异步、单向）

1.1：同步发送消息（每发送一条等待mq返回值）

1.2：异步发送消息

1.3：单向发送消息（不管成功失败，只管发送消息）

1.4：顺序发送消息

1.5：批量发送消息

1.6：定时发送消息

2：消费者

2.1：push消费

2.2：pull消费

---

1：生产者（同步、异步、单向）

在了解生产者之前，首先再次查看这个图片。生产者发送消息，围绕生产者的概念和怎么发送消息来解析MQ。生产者有重要的的message元素

Message：包含以下属性

字段名	默认值	必要性	说明
Topic	null	必填	消息所属 topic 的名称
Body	null	必填	消息体
Tags	null	选填	消息标签，方便服务器过滤使用。目前只支持每个消息设置一个
Keys	null	选填	代表这条消息的业务关键词，唯一ID
Flag	0	选填	完全由应用来设置，RocketMQ 不做干预
DelayTimeLevel	0	选填	消息延时级别，0 表示不延时，大于 0 会延时特定的时间才会被消费
WaitStoreMsgOK	true	选填	表示消息是否在服务器落盘后才返回应答。mq接受到消息，存入磁盘，然后返回成功或者失败

队列：为了支持高并发和水平扩展，需要对 Topic 进行分区，在 RocketMQ 中这被称为队列，一个 Topic 可能有多个队列，并且可能分布在不同的 Broker 上。保证消息的发送和消费的并发速度。在生产者将消息发送到MQ的broker的时候，这个时候broker的内部维护了队列，保证先进先出。默认一个topic里边有4个读4个写的队列

我们怎么发送消息，生产者发送消息包含同步，异步，单向这三个方面。当然按照功能又扩展出来顺序消息、批量消息、定时消息、事务消息等模式。

1.1：同步发送消息（每发送一条等待mq返回值）

同步发送：每次发送一条，等待mq的返回值成功，然后发送下一条，适合可靠的消息传递，适用范围最广泛。如重要的通知消息、短消息通知等

代码如下：

public class 普通消息发送_同步 {/*** 同步消息发送，发送之后等待服务端返回结果* 消息发送到broker 等待响应，成功后接着发送下一条数据* 保证了消息发送的可靠性** 使用场景：大部分可靠性要求高的场景*/public static void main(String[] args) throws MQClientException, MQBrokerException, RemotingException, InterruptedException {//1：初始化一个生产者，并且设置group是商品组System.out.println("同步发送消息模式");String producerGroup="shop_Group";DefaultMQProducer producer=new DefaultMQProducer(producerGroup);//2：设置nameServerproducer.setNamesrvAddr("localhost:9876");//3：启动producerproducer.start();//4：发送100条消息for (int i = 0; i < 10; i++) {//5:定义消息体Message msg=new Message();//设置消息主题 必填String topic="huyijuTopic";msg.setTopic(topic);//设置消息体 必填String body="同步:"+i;msg.setBody(body.getBytes(StandardCharsets.UTF_8));//设置落盘策略 默认落盘成功返回true 选填msg.setWaitStoreMsgOK(true);//设置消息keys 消息唯一标识 选填msg.setKeys("shop"+i);//设置消息标签 选填msg.setTags("同步");//6：发送数据SendResult sendResult = producer.send(msg);System.out.printf("%s%n", sendResult);}//7：发送完消息，关闭生产者producer.shutdown();}}

1.2：异步发送消息

异步发送消息：发送完后不用等待响应，就可以发送第二条消息，通过回调接口，来接受响应成功或者失败。异步发送一般用于链路耗时较长，对响应时间较为敏感的业务场景。例如，视频上传后通知启动转码服务，转码完成后通知推送转码结果等。

代码如下：

public class 普通消息发送_异步 {/*** 异步消息发送模式 发送数据之后不等响应接着发送，需要回调接口，回调接口告知失败 或者成功** 适用场景：适用于发送文件视频等大的文件  节省时间**/public static void main(String[] args) throws Exception {System.out.println("异步发送普通消息");// 初始化一个producer并设置Producer group nameDefaultMQProducer producer = new DefaultMQProducer("shop_Group");// 设置NameServer地址producer.setNamesrvAddr("localhost:9876");// 启动producerproducer.setRetryTimesWhenSendAsyncFailed(0);//重试次数producer.start();for (int i = 0; i < 5; i++) {// 创建一条消息，并指定topic、tag、body等信息，tag可以理解成标签，对消息进行再归类，RocketMQ可以在消费端对tag进行过滤String topic="huyijuTopic";Message msg = new Message(topic,"异步",("异步"+i).getBytes(StandardCharsets.UTF_8));// 异步发送消息, 发送结果通过callback返回给客户端producer.send(msg, new SendCallback() {@Overridepublic void onSuccess(SendResult sendResult) {System.out.println("数据发送成功："+sendResult);}@Overridepublic void onException(Throwable e) {System.out.println("数据发送失败："+e);e.printStackTrace();}});}// 一旦producer不再使用，关闭producerproducer.shutdown();}}

1.3：单向发送消息（不管成功失败，只管发送消息）

单向发送：生产者向mq发送消息，不等待mq的返回值，是否消息接收成功或者失败。生产者只管发送，适用于日志等可靠性不高的场景。发送速度很快，微秒级的速度

public class 普通消息发送_单向 {/*** 单向发送消息模式* 服务方只发送消息 不等服务端响应 也不管回调 发送速度很快 就是只管发送消息  不管成功失败** 适用场景：适用于发送日志，对于数据可靠性要求不高*/public static void main(String[] args) throws Exception {System.out.println("单向发送普通消息");// 初始化一个producer并设置Producer group nameDefaultMQProducer producer = new DefaultMQProducer("shop_Group");// 设置NameServer地址producer.setNamesrvAddr("127.0.0.1:9876");// 启动producerproducer.start();for (int i = 0; i < 10; i++) {final int index = i;// 创建一条消息，并指定topic、tag、body等信息，tag可以理解成标签，对消息进行再归类，RocketMQ可以在消费端对tag进行过滤String body="单向"+i;Message msg = new Message("huyijuTopic","单向",body.getBytes(StandardCharsets.UTF_8));// 不管不顾，直接发送，没有返回值producer.sendOneway(msg);}// 一旦producer不再使用，关闭producerproducer.shutdown();}}

1.4：顺序发送消息

我们知道，我们发送的消息，存储到了mq的topic的队列里边，默认的topic是4个队列

根据消息的key将消息轮训的插入队列中，队列的消息能保证FIFO，但是我们并不知道实际具体那条消息在那个队列，无法保证比如订单号是01的所有操作在同一个队列。如下图

 消费者再消费的时候，无法保证业务的一致性。所有才有了顺序发送，我们传入指定的订单号，只要订单号一直，就一定会存到相同的队列。

 代码如下：

/*** 顺序发送：SendResult send(Message msg, MessageQueueSelector selector, Object arg)* 根据同一个arg的值存入，相同的队列  队列一共四个，很多数据的时候，根据arg%4 存入指定的队列，先进先出** 适用场景：下单，支付，物流等场景，我们使用orderId作为分区id 会发送到同一个队列 保证顺序** 注意事项：只能有一个生产者，因为分布式环境，多个生产者发送相同的同一个orderId,无法判定先后顺序* 必须是单一的生产者** 如果一个Broker掉线，那么此时队列总数是否会发化？** 如果发生变化，那么同一个 ShardingKey 的消息就会发送到不同的队列上，造成乱序。* 如果不发生变化，那消息将会发送到掉线Broker的队列上，必然是失败的。因此 Apache RocketMQ 提供了两种模式，* 如果要保证严格顺序而不是可用性，创建 Topic 是要指定 -o 参数（--order）为true，表示顺序消息:** sh bin/mqadmin updateTopic -c DefaultCluster -t TopicTest -o true -n 127.0.0.1:9876*/
public class 顺序消息发送 {public static void main(String[] args) {System.out.println("顺序消息发送");DefaultMQProducer producer = new DefaultMQProducer("shop_Group");try {// 设置NameServer地址producer.setNamesrvAddr("127.0.0.1:9876");int a=producer.getDefaultTopicQueueNums();System.out.println("默认的队列大小："+a);producer.start();for (int i = 0; i < 5; i++) {
//                String[] tags = new String[] {"TagA", "TagB", "TagC", "TagD", "TagE"};
//                Message msg1 =
//                        new Message("TopicTest", tags[i % tags.length], "KEY" + i,
//                                ("Hello RocketMQ " + i).getBytes(RemotingHelper.DEFAULT_CHARSET));Message msg=new Message();//设置主题String topic="huyijuTopic";msg.setTopic(topic);//设置内容String body="顺序发送"+i;msg.setBody(body.getBytes(RemotingHelper.DEFAULT_CHARSET));//设置keysmsg.setKeys("key"+i);//设置tagsmsg.setTags("顺序发送");//订单id 根据不同的id将消息发送到不同的队列(队列总共4个 取模放入队列) 遵循FIFO
//                int orderId = i%a;
//                System.out.println("订单id:"+orderId);SendResult sendResult = producer.send(msg, new MessageQueueSelector() {@Overridepublic MessageQueue select(List<MessageQueue> list, Message message, Object arg) {
//                        System.out.println("参数arg:"+arg);//这里参数的send方法的参数一致
//                        System.out.println("参数list:"+list.size());//默认队列是4
//                        System.out.println(message);//根据订单id取模，存入指定的队列 然后返回该队列Integer id = (Integer) arg;int index = id % list.size();MessageQueue messageQueue = list.get(index);return messageQueue;}}, 5);//这里的5就是实际上我们的订单号，根据这个参数将消息存到相同的队列System.out.printf("%s%n", sendResult);}} catch (MQClientException e) {e.printStackTrace();} catch (UnsupportedEncodingException e) {e.printStackTrace();} catch (MQBrokerException e) {e.printStackTrace();} catch (RemotingException e) {e.printStackTrace();} catch (InterruptedException e) {e.printStackTrace();}finally {producer.shutdown();}}
}

1.5：批量发送消息

批量发送消息：将消息批量发送到mq来节省时间

代码如下：

/*** 批量投送消息，增加吞吐率 减少网络调用次数** 需要注意的是批量消息的大小不能超过 1MB**/
public class 批量消息发送 {public static void main(String[] args) throws MQClientException, MQBrokerException, RemotingException, InterruptedException {System.out.println("批量消息投送");DefaultMQProducer producer = new DefaultMQProducer("shop_Group");producer.setNamesrvAddr("127.0.0.1:9876");producer.start();String body1="批量任务1";String body2="批量任务2";String body3="批量任务3";Message message1=new Message("huyijuTopic","批量任务",body1.getBytes(StandardCharsets.UTF_8));Message message2=new Message("huyijuTopic","批量任务",body2.getBytes(StandardCharsets.UTF_8));Message message3=new Message("huyijuTopic","批量任务",body3.getBytes(StandardCharsets.UTF_8));List<Message> list=new ArrayList<>();list.add(message1);list.add(message2);list.add(message3);SendResult send = producer.send(list);System.out.println("批量消息投送结束"+send);producer.shutdown();}
}

1.6：定时发送消息

定时消息发送：将消息发送到mq，mq根据定时将消息发送给消费者。切记不要搞反了，消息是发送到mq之后，定时发送个消费者。

代码如下：

/*** 延时消息发送，数据发送到mq之后 指定的时间之后才能消费** 适用场景：定时任务、超时精准投送** 缺点：大量的定时任务 容易造成消息积压  时间一到 消费者亚历山大**/
public class 延时消息发送 {public static void main(String[] args) throws MQClientException, MQBrokerException, RemotingException, InterruptedException {DefaultMQProducer producer = new DefaultMQProducer("shop_Group");// 设置NameServer地址
//        1	1s	10	6min
//        2	5s	11	7min
//        3	10s	12	8min
//        4	30s	13	9min
//        5	1min	14	10minproducer.setNamesrvAddr("127.0.0.1:9876");producer.start();for (int i = 0; i < 10; i++) {String body= "定时任务"+i;Message message=new Message("huyijuTopic","定时任务",body.getBytes(StandardCharsets.UTF_8));message.setDelayTimeLevel(3);//设置的定时任务级别SendResult send = producer.send(message);System.out.println("定时消息返回值："+send);}producer.shutdown();}
}

2：消费者

消息的消费者很简单，只两种模式

第一种（推送模式）：订阅mq服务的topic，mq收到消息把消息推送给消费者，适用范围广

第二种（拉取模式）：订阅mq服务的topic，mq收到消息，消费者定时去mq拉取消息

2.1：push消费

普通消息的推送消费

//适用于普通的消息推送，不适合用于顺序消息的消费
public class 消息接收_推送1 {public static void main(String[] args) throws MQClientException {String group = "Shop_Group_push";//1:初始化消息接收组DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(group);//2:设置NameServer地址consumer.setNamesrvAddr("localhost:9876");//3:订阅一个或多个topic，并指定tag过滤条件，这里指定*表示接收所有tag的消息String topic="huyijuTopic";consumer.subscribe(topic,"*");consumer.setMessageModel(MessageModel.CLUSTERING);//默认是集群模式//consumer.setMessageModel(MessageModel.BROADCASTING);//这里是广播模式//4.1:注册回调接口来处理从Broker中收到的消息 单个对列保证先进先出//但是多个队列 被消费者并发消费   不能保证消费的顺序性consumer.registerMessageListener(new MessageListenerConcurrently() {@Overridepublic ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {System.out.println(list);for (int i = 0; i < list.size(); i++) {byte[] body = list.get(i).getBody();String resault= null;try {resault = new String(body,"utf-8");} catch (UnsupportedEncodingException e) {e.printStackTrace();}System.out.println("接受huyijuTopic的第"+i+"条消息："+resault);}// 返回消息消费状态，ConsumeConcurrentlyStatus.CONSUME_SUCCESS为消费成功return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;}});consumer.start();
//        consumer.shutdown();System.out.println("推送消息接收启动1");}
}

顺序消息的推送消费 

public class 消息接收_顺序消费1 {public static void main(String[] args) throws MQClientException {System.out.println("顺序消费1");String group = "Shop_Group_push";//1:初始化消息接收组DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(group);//2:设置NameServer地址consumer.setNamesrvAddr("localhost:9876");//3:订阅一个或多个topic，并指定tag过滤条件，这里指定*表示接收所有tag的消息String topic="huyijuTopic";consumer.subscribe(topic,"*");consumer.setMessageModel(MessageModel.CLUSTERING);//默认是集群模式//consumer.setMessageModel(MessageModel.BROADCASTING);//这里是广播模式//4.2:注册回调接口来处理从Broker中收到的消息 单个对列保证先进先出//但是多个队列 被消费者并发消费,不能保证消费的顺序性 这里使用MessageListenerOrderlyconsumer.registerMessageListener(new MessageListenerOrderly() {@Overridepublic ConsumeOrderlyStatus consumeMessage(List<MessageExt> list, ConsumeOrderlyContext consumeOrderlyContext) {//AtomicLong consumeTimes = new AtomicLong(0);consumeTimes.incrementAndGet();for (int i = 0; i < list.size(); i++) {byte[] body = list.get(i).getBody();String resault= null;try {resault = new String(body,"utf-8");} catch (UnsupportedEncodingException e) {e.printStackTrace();}System.out.println("接受huyijuTopic的第"+i+"条消息："+resault);}//返回消费状态return ConsumeOrderlyStatus.SUCCESS;}});consumer.start();
//        consumer.shutdown();System.out.println("推送消息接收启动1");}
}

2.2：pull消费

消费者一直循环去mq拉取消息

public class 消息接收_拉取消息2 {public static void main(String[] args) throws MQClientException {System.out.println("开始消息拉取");DefaultLitePullConsumer defaultLitePullConsumer = new DefaultLitePullConsumer();defaultLitePullConsumer.setConsumerGroup("Shop_Pull_Group");//订阅主题  拉取消息String topic="huyijuTopic";defaultLitePullConsumer.subscribe(topic, "*");defaultLitePullConsumer.setPullBatchSize(1);defaultLitePullConsumer.setNamesrvAddr("localhost:9876");defaultLitePullConsumer.start();try {while (true) {List<MessageExt> messageExts = defaultLitePullConsumer.poll();System.out.printf("%s%n", messageExts);System.out.println("拉取数据长度："+messageExts.size());for (int i = 0; i < messageExts.size(); i++) {byte[] body = messageExts.get(i).getBody();String resault= new String(body,"utf-8");System.out.println("拉取消息："+resault);}}} catch (UnsupportedEncodingException e) {e.printStackTrace();} finally {defaultLitePullConsumer.shutdown();}}
}

以上就是生产者和消费者的消息模型。