SpringBoot整合rockerMQ消息队列的方法

这篇文章主要介绍“SpringBoot整合rockerMQ消息队列的方法”，在日常操作中，相信很多人在SpringBoot整合rockerMQ消息队列的方法问题上存在疑惑，小编查阅了各式资料，整理出简单好用的操作方法，希望对大家解答”SpringBoot整合rockerMQ消息队列的方法”的疑惑有所帮助！接下来，请跟着小编一起来学习吧！1、maven依赖2、yml配置文件rocketmq:
###连接地址nameServer
name-server: www.kaicostudy.com:9876;
producer:
group: kaico_producer
server:
port: 80883、生产者4、消费者集群消费
当 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是TransactionMQProduc免费云主机域名er，则发送的是事务消息。如果节点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相互之间是没有影响的。同步： 发送消息后需等待结果，消息的可靠性高发送速度慢；异步： 消息发送后，回调通知结果，消息发送速度快，消息可靠性低；单向(oneway)：消息发送后，不关心结果，发送速度最快，消息可靠性最差，适用于在大量日志数据和用户行为数据等场景发送数据。主要三个步骤
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_FLUSH3、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 拉取下一批消息。Consumer 重试
Consumer 消费失败，这里有 3 种情况：返回 RECONSUME_LATER返回 null抛出异常Broker 收到这个响应后，会把这条消息放入重试队列，重新发送给 Consumer。注意：Broker 默认最多重试 16 次，如果重试 16 次都失败，就把这条消息放入死信队列，Consumer 可以订阅死信队列进行消费。重试只有在集群模式（MessageModel.CLUSTERING）下生效，在广播模式（MessageModel.BROADCASTING）下是不生效的。Consumer 端一定要做好幂等处理。生产者投递消息根据key投递到同一个队列中存放消费者应该订阅到同一个队列实现消费最终应该使用同一个线程去消费消息(不能够实现多线程消费。)生产者代码消费者代码实现思路生产者（发送方）投递事务消息到Broker中，设置该消息为半消息 不可以被消费；开始执行我们的本地事务，将本地事务执行的结果（回滚或者提交）发送给BrokerBroker获取回滚或者提交，如果是回滚的情况则删除该消息、如果是提交的话，该消息就可以被消费者消费；Broker如果没有及时的获取发送方本地事务结果的话，会主动查询本地事务结果。1、生产者发送事务消息sendMessageInTransaction2、事务监听类3、消费者消费消息到此，关于“SpringBoot整合rockerMQ消息队列的方法”的学习就结束了，希望能够解决大家的疑惑。理论与实践的搭配能更好的帮助大家学习，快去试试吧！若想继续学习更多相关知识，请继续关注百云主机网站，小编会继续努力为大家带来更多实用的文章！

相关推荐: 怎么利用Qt实现可扩展对话框

这篇文章主要介绍“怎么利用Qt实现可扩展对话框”的相关知识，小编通过实际案例向大家展示操作过程，操作方法简单快捷，实用性强，希望这篇“怎么利用Qt实现可扩展对话框”文章能帮助大家解决问题。可扩展对话框通常用于用户对界面有不同要求的场合。通常情况下，只出现在基本…