目录
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：设置nameServer
producer.setNamesrvAddr("localhost:9876");
//3：启动producer
producer.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 name
DefaultMQProducer producer = new DefaultMQProducer("shop_Group");
// 设置NameServer地址
producer.setNamesrvAddr("localhost:9876");
// 启动producer
producer.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() {
@Override
public void onSuccess(SendResult sendResult) {
System.out.println("数据发送成功："+sendResult);
}
@Override
public void onException(Throwable e) {
System.out.println("数据发送失败："+e);
e.printStackTrace();
}
});
}
// 一旦producer不再使用，关闭producer
producer.shutdown();
}
}
1.3：单向发送消息（不管成功失败，只管发送消息）
单向发送：生产者向mq发送消息，不等待mq的返回值，是否消息接收成功或者失败。生产者只管发送，适用于日志等可靠性不高的场景。发送速度很快，微秒级的速度
public class 普通消息发送_单向 {
/**
* 单向发送消息模式
* 服务方只发送消息 不等服务端响应 也不管回调 发送速度很快 就是只管发送消息 不管成功失败
*
* 适用场景：适用于发送日志，对于数据可靠性要求不高
*/
public static void main(String[] args) throws Exception {
System.out.println("单向发送普通消息");
// 初始化一个producer并设置Producer group name
DefaultMQProducer producer = new DefaultMQProducer("shop_Group");
// 设置NameServer地址
producer.setNamesrvAddr("127.0.0.1:9876");
// 启动producer
producer.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不再使用，关闭producer
producer.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));
//设置keys
msg.setKeys("key"+i);
//设置tags
msg.setTags("顺序发送");
//订单id 根据不同的id将消息发送到不同的队列(队列总共4个 取模放入队列) 遵循FIFO
// int orderId = i%a;
// System.out.println("订单id:"+orderId);
SendResult sendResult = producer.send(msg, new MessageQueueSelector() {
@Override
public 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 10min
producer.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() {
@Override
public 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中收到的消息 单个对列保证先进先出
//但是多个队列 被消费者并发消费,不能保证消费的顺序性 这里使用MessageListenerOrderly
consumer.registerMessageListener(new MessageListenerOrderly() {
@Override
public 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();
}
}
}
以上就是生产者和消费者的消息模型。
