SpringBoot整合rockerMQ
kaico2018 人气:0Springboot整合RockerMQ
1、maven依赖
<dependencies>
<!-- springboot-web组件 -->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>
<dependency>
<groupId>org.apache.rocketmq</groupId>
<artifactId>rocketmq-spring-boot-starter</artifactId>
<version>2.0.3</version>
</dependency>
<dependency>
<groupId>org.projectlombok</groupId>
<artifactId>lombok</artifactId>
</dependency>
</dependencies>
2、yml配置文件
rocketmq:
###连接地址nameServer
name-server: www.kaicostudy.com:9876;
producer:
group: kaico_producer
server:
port: 8088
3、生产者
@RequestMapping("/sendMsg")
public String sendMsg() {
OrderEntity orderEntity = new OrderEntity("123456","腾讯视频会员");
SendResult kaicoTopic = rocketMQTemplate.syncSend("kaicoTopic"+":"+"tag1", orderEntity);
System.out.println("返回发送消息状态：" + kaicoTopic);
return "success";
}
4、消费者
@Service
@RocketMQMessageListener(topic = "kaicoTopic", selectorExpression ="tag1", consumerGroup = "kaico_consumer", messageModel = MessageModel.CLUSTERING)
public class OrdeConsumer2 implements RocketMQListener<OrderEntity> {
@Override
public void onMessage(OrderEntity o) {
System.out.println("kaico_consumer2消费者接收对象:" + o.toString());
}
}
使用总结
消费模式
集群消费
当 consumer 使用集群消费时，每条消息只会被 consumer 集群内的任意一个 consumer 实例消费一次。
同时记住一点，使用集群消费的时候，consumer 的消费进度是存储在 broker 上，consumer 自身是不存储消费进度的。消息进度存储在 broker 上的好处在于，当你 consumer 集群是扩大或者缩小时，由于消费进度统一在broker上，消息重复的概率会被大大降低了。
注意： 在集群消费模式下，并不能保证每一次消息失败重投都投递到同一个 consumer 实例。
注解配置：messageModel = MessageModel.CLUSTERING
广播消费
当 consumer 使用广播消费时，每条消息都会被 consumer 集群内所有的 consumer 实例消费一次，也就是说每条消息至少被每一个 consumer 实例消费一次。
与集群消费不同的是，consumer 的消费进度是存储在各个 consumer 实例上，这就容易造成消息重复。还有很重要的一点，对于广播消费来说，是不会进行消费失败重投的，所以在 consumer 端消费逻辑处理时，需要额外关注消费失败的情况。
虽然广播消费能保证集群内每个 consumer 实例都能消费消息，但是消费进度的维护、不具备消息重投的机制大大影响了实际的使用。因此，在实际使用中，更推荐使用集群消费，因为集群消费不仅拥有消费进度存储的可靠性，还具有消息重投的机制。而且，我们通过集群消费也可以达到广播消费的效果。
注解配置：messageModel = MessageModel.BROADCASTING
生产者组和消费者组
生产者组
一个生产者组，代表着一群topic相同的Producer。即一个生产者组是同一类Producer的组合。
如果Producer是TransactionMQProducer，则发送的是事务消息。如果节点1发送完消息后，消息存储到broker的Half Message Queue中，还未存储到目标topic的queue中时，此时节点1崩溃，则可以通过同一Group下的节点2进行二阶段提交，或回溯。
使用时，一个节点下，一个topic会对应一个producer
消费者组
一个消费者组，代表着一群topic相同，tag相同（即逻辑相同）的Consumer。通过一个消费者组，则可容易的进行负载均衡以及容错
使用时，一个节点下，一个topic加一个tag可以对应一个consumer。一个消费者组就是横向上多个节点的相同consumer为一个消费组。
首先分析一下producer。习惯上我们不会创建多个订阅了相同topic的Producer实例，因为一个Producer实例发送消息时是通过ExecutorService线程池去异步执行的，不会阻塞完全够用，如果创建了多个相同topic的Producer则会影响性能。而Consumer则不同。消息会在一topic下会细分多个tag，需要针对tag需要针对不同的tag创建多个消费者实例。
注意：多个不同的消费者组订阅同一个topic、tag，如果设定的是集群消费模式，每一个消费者组中都会有一个消费者来消费。也就是说不同的消费者组订阅同一个topic相互之间是没有影响的。
生产者投递消息的三种方式
同步： 发送消息后需等待结果，消息的可靠性高发送速度慢；
SendResult kaicoTopic = rocketMQTemplate.syncSend("kaicoTopic"+":"+"tag1", orderEntity);
异步： 消息发送后，回调通知结果，消息发送速度快，消息可靠性低；
//异步发送
rocketMQTemplate.asyncSend("kaicoTopic" + ":" + "tag1", orderEntity, new SendCallback() {
@Override
public void onSuccess(SendResult sendResult) {
System.out.println("异步发送消息成功");
}
@Override
public void onException(Throwable throwable) {
System.out.println("异步发送消息失败");
}
});单向(oneway)：消息发送后，不关心结果，发送速度最快，消息可靠性最差，适用于在大量日志数据和用户行为数据等场景发送数据。
//单向(oneway)发送
rocketMQTemplate.sendOneWay("kaicoTopic"+":"+"tag1", orderEntity);如何保证消息不丢失
主要三个步骤
1、生产者保证消息发送成功
采用同步发送消息的方式，发送消息后有返回结果，保证消息发送成功。（代码见上面）
返回四个状态
- SEND_OK：消息发送成功。需要注意的是，消息发送到 broker 后，还有两个操作：消息刷盘和消息同步到 slave 节点，默认这两个操作都是异步的，只有把这两个操作都改为同步，SEND_OK 这个状态才能真正表示发送成功。
- FLUSH_DISK_TIMEOUT：消息发送成功但是消息刷盘超时。
- FLUSH_SLAVE_TIMEOUT：消息发送成功但是消息同步到 slave 节点时超时。
- SLAVE_NOT_AVAILABLE：消息发送成功但是 broker 的 slave 节点不可用。
2、rocketMQ将消息持久化，保证宕机后消息不会丢失。持久化策略（刷盘策略）
- 异步刷盘：默认。消息写入 CommitLog 时，并不会直接写入磁盘，而是先写入 PageCache 缓存后返回成功，然后用后台线程异步把消息刷入磁盘。异步刷盘提高了消息吞吐量，但是可能会有消息丢失的情况，比如断点导致机器停机，PageCache 中没来得及刷盘的消息就会丢失。
- 同步刷盘：消息写入内存后，立刻请求刷盘线程进行刷盘，如果消息未在约定的时间内(默认 5 s)刷盘成功，就返回 FLUSH_DISK_TIMEOUT，Producer 收到这个响应后，可以进行重试。同步刷盘策略保证了消息的可靠性，同时降低了吞吐量，增加了延迟。要开启同步刷盘，需要增加下面配置：
flushDiskType=SYNC_FLUSH
3、Broker 多副本和高可用
Broker 为了保证高可用，采用一主多从的方式部署。
消息发送到 master 节点后，slave 节点会从 master 拉取消息保持跟 master 的一致。这个过程默认是异步的，即 master 收到消息后，不等 slave 节点复制消息就直接给 Producer 返回成功。
这样会有一个问题，如果 slave 节点还没有完成消息复制，这时 master 宕机了，进行主备切换后就会有消息丢失。为了避免这个问题，可以采用 slave 节点同步复制消息，即等 slave 节点复制消息成功后再给 Producer 返回发送成功。只需要增加下面的配置：brokerRole=SYNC_MASTER
改为同步复制后，消息复制流程如下：
- slave 初始化后，跟 master 建立连接并向 master 发送自己的 offset；
- master 收到 slave 发送的 offset 后，将 offset 后面的消息批量发送给 slave；
- slave 把收到的消息写入 commitLog 文件，并给 master 发送新的 offset；
- master 收到新的 offset 后，如果 offset >= producer 发送消息后的 offset，给 Producer 返回 SEND_OK。
4、消费者保证消息消费成功
消费者消费消息后，如果 Consumer 消费成功，返回 CONSUME_SUCCESS，提交 offset 并从 Broker 拉取下一批消息。
@Service
public class NoSpringBootOrderConsumer {
private DefaultMQPushConsumer defaultMQPushConsumer;
@Value("${rocketmq.name-server}")
private String namesrvAddr;
protected String consumerGroup;
protected String topic;
protected String topicTag;
public void setNamesrvAddr(String namesrvAddr) {
this.namesrvAddr = namesrvAddr;
}
public void setConsumerGroup(String consumerGroup) {
this.consumerGroup = consumerGroup;
}
public void setTopic(String topic) {
this.topic = topic;
}
public void setTopicTag(String topicTag) {
this.topicTag = topicTag;
}
public static String encoding = System.getProperty("file.encoding");
/*
* @Author ex_fengkai
* @Description //TODO 初始化数据（消费者组名称、topic、topic的tag、nameServer的信息）
* @Date 2020/11/9 14:36
* @Param []
* @return void
**/
private void initParam() {
this.consumerGroup = "kaico_consumer3";
this.topic = "kaicoTopic";
this.topicTag = "tag1";
this.setNamesrvAddr(namesrvAddr);
}
@PostConstruct
private void init() throws InterruptedException, MQClientException {
initParam();
// ConsumerGroupName需要由应用来保证唯一，用于把多个Consumer组织到一起，提高并发处理能力
defaultMQPushConsumer = new DefaultMQPushConsumer(consumerGroup);
defaultMQPushConsumer.setNamesrvAddr(namesrvAddr); //设置nameServer服务器
defaultMQPushConsumer.setInstanceName(String.valueOf(System.currentTimeMillis()));
defaultMQPushConsumer.setVipChannelEnabled(false);
// 设置Consumer第一次启动是从队列头部开始消费
defaultMQPushConsumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
// 订阅指定Topic下的topicTag
System.out.println("consumerGroup:" + consumerGroup + " topic:" + topic + " ,topicTag:" + topicTag);
defaultMQPushConsumer.subscribe(topic, topicTag);
// 设置为集群消费
defaultMQPushConsumer.setMessageModel(MessageModel.CLUSTERING);
// 通过匿名消息监听处理消息消费
defaultMQPushConsumer.registerMessageListener(new MessageListenerConcurrently() {
// 默认msgs里只有一条消息，可以通过设置consumeMessageBatchMaxSize参数来批量接收消息
@Override
public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
MessageExt msg = msgs.get(0);
if (msg.getTopic().equals(topic) && msg.getTags() != null && msg.getTags().equals(topicTag)) {
// 执行topic下对应tag的消费逻辑
try {
onMessage(new String(msg.getBody(),"utf-8"));
} catch (UnsupportedEncodingException e) {
System.out.println("系统不支持消息编码格式:" + encoding);
return ConsumeConcurrentlyStatus.RECONSUME_LATER;
} catch (Exception e) {
System.out.println("消息处理异常");
return ConsumeConcurrentlyStatus.RECONSUME_LATER;
}
System.out.println("consumerGroup:" + consumerGroup + " MsgId:" + msg.getMsgId() + " was done!");
}
// 如果没有return success ，consumer会重新消费该消息，直到return success
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
}
});
// Consumer对象在使用之前必须要调用start初始化，初始化一次即可
defaultMQPushConsumer.start();
System.out.println("consumerGroup:" + consumerGroup + " namesrvAddr:" + namesrvAddr + " start success!");
}
@PreDestroy
public void destroy() {
defaultMQPushConsumer.shutdown();
}
private void onMessage(String s) {
System.out.println(consumerGroup + "用spring的方式的消费者消费：" + s);
}
}Consumer 重试
Consumer 消费失败，这里有 3 种情况：
- 返回 RECONSUME_LATER
- 返回 null
- 抛出异常
Broker 收到这个响应后，会把这条消息放入重试队列，重新发送给 Consumer。
注意：Broker 默认最多重试 16 次，如果重试 16 次都失败，就把这条消息放入死信队列，Consumer 可以订阅死信队列进行消费。重试只有在集群模式（MessageModel.CLUSTERING）下生效，在广播模式（MessageModel.BROADCASTING）下是不生效的。Consumer 端一定要做好幂等处理。
顺序消息
- 生产者投递消息根据key投递到同一个队列中存放
- 消费者应该订阅到同一个队列实现消费
- 最终应该使用同一个线程去消费消息(不能够实现多线程消费。)
生产者代码
//发送顺序消息
@RequestMapping("/sendMsg1")
public String sendMsg1() throws InterruptedException, RemotingException, MQClientException, MQBrokerException {
Long orderId = System.currentTimeMillis();
String insertSql = getSqlMsg("insert", orderId);
String updateSql = getSqlMsg("update", orderId);
String deleteSql = getSqlMsg("delete", orderId);
Message insertMsg = new Message("kaicoTopic", "tag6", insertSql.getBytes());
Message updateMsg = new Message("kaicoTopic", "tag6", updateSql.getBytes());
Message deleteMsg = new Message("kaicoTopic", "tag6", deleteSql.getBytes());
DefaultMQProducer producer = rocketMQTemplate.getProducer();
rocketMQTemplate.getProducer().send(insertMsg
, new MessageQueueSelector() {
@Override
public MessageQueue select(List<MessageQueue> mqs, Message msg,
Object arg) {
// 该消息存放到队列0中
return mqs.get(0);
}
}, orderId);
rocketMQTemplate.getProducer().send(updateMsg
, new MessageQueueSelector() {
@Override
public MessageQueue select(List<MessageQueue> mqs, Message msg,
Object arg) {
// 该消息存放到队列0中
return mqs.get(0);
}
}, orderId);
rocketMQTemplate.getProducer().send(deleteMsg
, new MessageQueueSelector() {
@Override
public MessageQueue select(List<MessageQueue> mqs, Message msg,
Object arg) {
// 该消息存放到队列0中
return mqs.get(0);
}
}, orderId);
return orderId + "";
}消费者代码
@Service
@RocketMQMessageListener(topic = "kaicoTopic", selectorExpression ="tag6", consumerGroup = "kaico_consumer1",
messageModel = MessageModel.CLUSTERING, consumeMode = ConsumeMode.ORDERLY, consumeThreadMax = 1)
public class OrdeConsumer implements RocketMQListener<MessageExt> {
@Override
public void onMessage(MessageExt msg) {
System.out.println(Thread.currentThread().getName() + "-kaico_consumer1消费者接收对象:队列" + msg.getQueueId()
+ "=消息：" + new String(msg.getBody()));
}
}分布式事务
实现思路
- 生产者（发送方）投递事务消息到Broker中，设置该消息为半消息 不可以被消费；
- 开始执行我们的本地事务，将本地事务执行的结果（回滚或者提交）发送给Broker
- Broker获取回滚或者提交，如果是回滚的情况则删除该消息、如果是提交的话，该消息就可以被消费者消费；
- Broker如果没有及时的获取发送方本地事务结果的话，会主动查询本地事务结果。
1、生产者发送事务消息sendMessageInTransaction
public String saveOrder() {
// 提前生成我们的订单id
String orderId = System.currentTimeMillis() + "";
/**
* 1.提前生成我们的半消息
* 2.半消息发送成功之后，在执行我们的本地事务
*/
OrderEntity orderEntity = createOrder(orderId);
String msg = JSONObject.toJSONString(orderEntity);
MessageBuilder<String> stringMessageBuilder = MessageBuilder.withPayload(msg);
stringMessageBuilder.setHeader("msg", msg);
Message message = stringMessageBuilder.build();
// 该消息不允许被消费者消费，生产者的事务逻辑代码在生产者事务监听类中executeLocalTransaction方法中执行。
rocketMQTemplate.sendMessageInTransaction("kaicoProducer",
"orderTopic", message, null);
return orderId;
}2、事务监听类
@Slf4j
@Component
@RocketMQTransactionListener(txProducerGroup = "kaicoProducer") //这个mayiktProducer生产者的事务管理
public class SyncProducerListener implements RocketMQLocalTransactionListener {
@Autowired
private OrderMapper orderMapper;
@Autowired
private TransationalUtils transationalUtils;
/**
* 执行我们订单的事务
* @param msg
* @param arg
* @return
*/
@Override
public RocketMQLocalTransactionState executeLocalTransaction(Message msg, Object arg) {
MessageHeaders headers = msg.getHeaders();
//拿到消息
Object object = headers.get("msg");
if (object == null) {
return null;
}
String orderMsg = (String) object;
OrderEntity orderEntity = JSONObject.parseObject(orderMsg, OrderEntity.class);
TransactionStatus begin = null;
try {
begin = transationalUtils.begin();
int result = orderMapper.addOrder(orderEntity);
transationalUtils.commit(begin);
if (result <= 0) {
return RocketMQLocalTransactionState.ROLLBACK;
}
// 告诉我们的Broke可以消费者该消息
return RocketMQLocalTransactionState.COMMIT;
} catch (Exception e) {
if (begin != null) {
transationalUtils.rollback(begin);
return RocketMQLocalTransactionState.ROLLBACK;
}
}
//add.Order
return null;
}
/**
* 提供给我们的Broker定时检查
* @param msg
* @return
*/
@Override
public RocketMQLocalTransactionState checkLocalTransaction(Message msg) {
MessageHeaders headers = msg.getHeaders();
Object object = headers.get("msg");
if (object == null) {
return RocketMQLocalTransactionState.ROLLBACK;
}
String orderMsg = (String) object;
OrderEntity orderEntity = JSONObject.parseObject(orderMsg, OrderEntity.class);
String orderId = orderEntity.getOrderId();
// 直接查询我们的数据库
OrderEntity orderDbEntity = orderMapper.findOrderId(orderId);
if (orderDbEntity == null) {
//不确认，继续重试
return RocketMQLocalTransactionState.UNKNOWN;
}
//提交事务
return RocketMQLocalTransactionState.COMMIT;
}
}3、消费者消费消息
@Service
@RocketMQMessageListener(topic = "orderTopic", consumerGroup = "kaicoTopic")
public class OrdeConsumer implements RocketMQListener<String> {
@Autowired
private DispatchMapper dispatchMapper;
@Override
public void onMessage(String msg) {
OrderEntity orderEntity = JSONObject.parseObject(msg, OrderEntity.class);
String orderId = orderEntity.getOrderId();
// 模拟userid为=123456
DispatchEntity dispatchEntity = new DispatchEntity(orderId, 123456L);
dispatchMapper.insertDistribute(dispatchEntity);
}
}
加载全部内容