第六章 RocketMQ应用（一）

<properties>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
<maven.compiler.source>1.8</maven.compiler.source>
<maven.compiler.target>1.8</maven.compiler.target>
</properties>
<dependencies>
<dependency>
<groupId>org.apache.rocketmq</groupId>
<artifactId>rocketmq-client</artifactId>
<version>4.8.0</version>
</dependency>
</dependencies>

public class SyncProducer {
public static void main(String[] args) throws Exception {
// 创建一个producer，参数为Producer Group名称
DefaultMQProducer producer = new DefaultMQProducer("pg");
// 指定nameServer地址
producer.setNamesrvAddr("rocketmqOS:9876");
// 设置当发送失败时重试发送的次数，默认为2次
producer.setRetryTimesWhenSendFailed(3);
// 设置发送超时时限为5s，默认3s
producer.setSendMsgTimeout(5000);
// 开启生产者
producer.start();
// 生产并发送100条消息
for (int i = 0; i < 100; i++) {
byte[] body = ("Hi," + i).getBytes();
Message msg = new Message("someTopic", "someTag", body);
// 为消息指定key
msg.setKeys("key-" + i);
// 发送消息
SendResult sendResult = producer.send(msg);
System.out.println(sendResult);
}
// 关闭producer
producer.shutdown();
}
}

// 消息发送的状态
public enum SendStatus {
SEND_OK, // 发送成功
FLUSH_DISK_TIMEOUT, // 刷盘超时。当Broker设置的刷盘策略为同步刷盘时才可能出
现这种异常状态。异步刷盘不会出现
FLUSH_SLAVE_TIMEOUT, // Slave同步超时。当Broker集群设置的Master-Slave的复
制方式为同步复制时才可能出现这种异常状态。异步复制不会出现
SLAVE_NOT_AVAILABLE, // 没有可用的Slave。当Broker集群设置为Master-Slave的
复制方式为同步复制时才可能出现这种异常状态。异步复制不会出现
}   

public class AsyncProducer {
public static void main(String[] args) throws Exception {
DefaultMQProducer producer = new DefaultMQProducer("pg");
producer.setNamesrvAddr("rocketmqOS:9876");
// 指定异步发送失败后不进行重试发送
producer.setRetryTimesWhenSendAsyncFailed(0);
// 指定新创建的Topic的Queue数量为2，默认为4
producer.setDefaultTopicQueueNums(2);
producer.start();
for (int i = 0; i < 100; i++) {
byte[] body = ("Hi," + i).getBytes();
try {
Message msg = new Message("myTopicA", "myTag", body);
// 异步发送。指定回调
producer.send(msg, new SendCallback() {
// 当producer接收到MQ发送来的ACK后就会触发该回调方法的执行
@Override
public void onSuccess(SendResult sendResult) {
System.out.println(sendResult);
}
@Override
public void onException(Throwable e) {
e.printStackTrace();
}
});
} catch (Exception e) {
e.printStackTrace();
}
} // end-for
// sleep一会儿
// 由于采用的是异步发送，所以若这里不sleep，
// 则消息还未发送就会将producer给关闭，报错
TimeUnit.SECONDS.sleep(3);
producer.shutdown();
}
}

public class OnewayProducer {
public static void main(String[] args) throws Exception{
DefaultMQProducer producer = new DefaultMQProducer("pg");
producer.setNamesrvAddr("rocketmqOS:9876");
producer.start();
for (int i = 0; i < 10; i++) {
byte[] body = ("Hi," + i).getBytes();
Message msg = new Message("single", "someTag", body);
// 单向发送
producer.sendOneway(msg);
}
producer.shutdown();
System.out.println("producer shutdown");
}
}  

public class SomeConsumer {
public static void main(String[] args) throws MQClientException {
// 定义一个pull消费者
// DefaultLitePullConsumer consumer = new
DefaultLitePullConsumer("cg");
// 定义一个push消费者
DefaultMQPushConsumer consumer = new
DefaultMQPushConsumer("cg");
// 指定nameServer
consumer.setNamesrvAddr("rocketmqOS:9876");
// 指定从第一条消息开始消费
consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET
);
// 指定消费topic与tag
consumer.subscribe("someTopic", "*");
// 指定采用“广播模式”进行消费，默认为“集群模式”
// consumer.setMessageModel(MessageModel.BROADCASTING);
// 注册消息监听器
consumer.registerMessageListener(new
MessageListenerConcurrently() {
// 一旦broker中有了其订阅的消息就会触发该方法的执行，
// 其返回值为当前consumer消费的状态
@Override
public ConsumeConcurrentlyStatus
consumeMessage(List<MessageExt> msgs,
ConsumeConcurrentlyContext context) {
// 逐条消费消息
for (MessageExt msg : msgs) {
System.out.println(msg);
}
// 返回消费状态：消费成功
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
}
});
// 开启消费者消费
consumer.start();
System.out.println("Consumer Started");
}
}

在创建Topic时指定Queue的数量。有三种指定方式：
1）在代码中创建Producer时，可以指定其自动创建的Topic的Queue数量
2）在RocketMQ可视化控制台中手动创建Topic时指定Queue数量
3）使用mqadmin命令手动创建Topic时指定Queue数量

如何实现Queue的选择？在定义Producer时我们可以指定消息队列选择器，而这个选择器是我们
自己实现了MessageQueueSelector接口定义的。
在定义选择器的选择算法时，一般需要使用选择key。这个选择key可以是消息key也可以是其它
数据。但无论谁做选择key，都不能重复，都是唯一的。

一般性的选择算法是，让选择key（或其hash值）与该Topic所包含的Queue的数量取模，其结果
即为选择出的Queue的QueueId。
取模算法存在一个问题：不同选择key与Queue数量取模结果可能会是相同的，即不同选择key的
消息可能会出现在相同的Queue，即同一个Consuemr可能会消费到不同选择key的消息。这个问
题如何解决？一般性的作法是，从消息中获取到选择key，对其进行判断。若是当前Consumer需
要消费的消息，则直接消费，否则，什么也不做。这种做法要求选择key要能够随着消息一起被
Consumer获取到。此时使用消息key作为选择key是比较好的做法。
以上做法会不会出现如下新的问题呢？不属于那个Consumer的消息被拉取走了，那么应该消费
该消息的Consumer是否还能再消费该消息呢？同一个Queue中的消息不可能被同一个Group中的
不同Consumer同时消费。所以，消费现一个Queue的不同选择key的消息的Consumer一定属于不
同的Group。而不同的Group中的Consumer间的消费是相互隔离的，互不影响的。  

public class OrderedProducer {
public static void main(String[] args) throws Exception {
DefaultMQProducer producer = new DefaultMQProducer("pg");
producer.setNamesrvAddr("rocketmqOS:9876");
producer.start();
for (int i = 0; i < 100; i++) {
Integer orderId = i;
byte[] body = ("Hi," + i).getBytes();
Message msg = new Message("TopicA", "TagA", body);
SendResult sendResult = producer.send(msg, new
MessageQueueSelector() {
@Override
public MessageQueue select(List<MessageQueue> mqs,
Message msg, Object arg) {
Integer id = (Integer) arg;
int index = id % mqs.size();
return mqs.get(index);
}
}, orderId);
System.out.println(sendResult);
}
producer.shutdown();
}
}

在电商平台中，订单创建时会发送一条延迟消息。这条消息将会在30分钟后投递给后台业务系
统（Consumer），后台业务系统收到该消息后会判断对应的订单是否已经完成支付。如果未完
成，则取消订单，将商品再次放回到库存；如果完成支付，则忽略。
在12306平台中，车票预订成功后就会发送一条延迟消息。这条消息将会在45分钟后投递给后台
业务系统（Consumer），后台业务系统收到该消息后会判断对应的订单是否已经完成支付。如
果未完成，则取消预订，将车票再次放回到票池；如果完成支付，则忽略。

messageDelayLevel = 1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h 1d  

延迟等级delayLevel与queueId的对应关系为queueId = delayLevel -1
需要注意，在创建queueId目录时，并不是一次性地将所有延迟等级对应的目录全部创建完毕，
而是用到哪个延迟等级创建哪个目录

SCHEDULE_TOPIC_XXXX目录中各个延时等级Queue中的消息是如何排序的？
是按照消息投递时间排序的。一个Broker中同一等级的所有延时消息会被写入到consumequeue
目录中SCHEDULE_TOPIC_XXXX目录下相同Queue中。即一个Queue中消息投递时间的延迟等
级时间是相同的。那么投递时间就取决于于消息存储时间了。即按照消息被发送到Broker的时
间进行排序的。  

ScheuleMessageService在Broker启动时，会创建并启动一个定时器TImer，用于执行相应的定时
任务。系统会根据延时等级的个数，定义相应数量的TimerTask，每个TimerTask负责一个延迟
等级消息的消费与投递。每个TimerTask都会检测相应Queue队列的第一条消息是否到期。若第
一条消息未到期，则后面的所有消息更不会到期（消息是按照投递时间排序的）；若第一条消
息到期了，则将该消息投递到目标Topic，即消费该消息。  

这其实就是一次普通消息发送。只不过这次的消息Producer是延迟消息服务类
ScheuleMessageService。   

public class DelayProducer {
public static void main(String[] args) throws Exception {
DefaultMQProducer producer = new DefaultMQProducer("pg");
producer.setNamesrvAddr("rocketmqOS:9876");
producer.start();
for (int i = 0; i < 10; i++) {
byte[] body = ("Hi," + i).getBytes();
Message msg = new Message("TopicB", "someTag", body);
// 指定消息延迟等级为3级，即延迟10s
// msg.setDelayTimeLevel(3);
SendResult sendResult = producer.send(msg);
// 输出消息被发送的时间
System.out.print(new SimpleDateFormat("mm:ss").format(new
Date()));
System.out.println(" ," + sendResult);
}
producer.shutdown();
}
}

public class OtherConsumer {
public static void main(String[] args) throws MQClientException {
DefaultMQPushConsumer consumer = new
DefaultMQPushConsumer("cg");
consumer.setNamesrvAddr("rocketmqOS:9876");
consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET
);
consumer.subscribe("TopicB", "*");
consumer.registerMessageListener(new
MessageListenerConcurrently() {
@Override
public ConsumeConcurrentlyStatus
consumeMessage(List<MessageExt> msgs,
ConsumeConcurrentlyContext context) {
for (MessageExt msg : msgs) {
// 输出消息被消费的时间
System.out.print(new
SimpleDateFormat("mm:ss").format(new Date()));
System.out.println(" ," + msg);
}
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
}
});
consumer.start();
System.out.println("Consumer Started");
}
}