亲宝软件园·资讯

展开

SpringBoot整合rockerMQ

kaico2018 人气:0

Springboot整合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、生产者保证消息发送成功
采用同步发送消息的方式,发送消息后有返回结果,保证消息发送成功。(代码见上面)
返回四个状态

2、rocketMQ将消息持久化,保证宕机后消息不会丢失。持久化策略(刷盘策略)

flushDiskType=SYNC_FLUSH

3、Broker 多副本和高可用
Broker 为了保证高可用,采用一主多从的方式部署。
消息发送到 master 节点后,slave 节点会从 master 拉取消息保持跟 master 的一致。这个过程默认是异步的,即 master 收到消息后,不等 slave 节点复制消息就直接给 Producer 返回成功。

这样会有一个问题,如果 slave 节点还没有完成消息复制,这时 master 宕机了,进行主备切换后就会有消息丢失。为了避免这个问题,可以采用 slave 节点同步复制消息,即等 slave 节点复制消息成功后再给 Producer 返回发送成功。只需要增加下面的配置:
brokerRole=SYNC_MASTER

改为同步复制后,消息复制流程如下:

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 种情况:

Broker 收到这个响应后,会把这条消息放入重试队列,重新发送给 Consumer。

注意:Broker 默认最多重试 16 次,如果重试 16 次都失败,就把这条消息放入死信队列,Consumer 可以订阅死信队列进行消费。重试只有在集群模式(MessageModel.CLUSTERING)下生效,在广播模式(MessageModel.BROADCASTING)下是不生效的。Consumer 端一定要做好幂等处理。

顺序消息

生产者代码

//发送顺序消息
    @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()));
    }
}

分布式事务

在这里插入图片描述

实现思路

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);
    }
}

加载全部内容

相关教程
猜你喜欢
用户评论