什么是MQ

消息队列Message Queue简称MQ是在消息的传输过程中保存消息的容器用于分布式系统之间进行通信。

 MQ的选型和对比

 在讲RabbitMQ之前先说一下AMQP即 Advanced Message Queuing Protocol(高级消息队列协议)是一个网络协议是应用层协议的一个开放标准为面向消息的中间件设计。基于此协议的客户端与消息中间件可传递消息并不受客户端/中间件不同产品不同的开发语言等条件的限制。2006年AMQP规范发布。类比HTTP。

其架构如图

 从这个图可以了解到RabbitMQ的四个重要接口

connection连接
channel轻量级的connection信道核心处理大部分的API实现
exchange只负责分发消息若没有queue绑定到exchange上则消息会丢失
queue存储消息的容器

 2007年Rabbit技术公司基于AMQP标准开发的RabbitMQ1.0发布。RabbitMQ采用Erlang 语言开发。

RabbitMQ

使用RabbitMQ的三大好处

解耦

传统系统之间的耦合度太强主要集中在两部分

业务之间的耦合度和系统之间的耦合度。

比如说存在这样一个微服务场景A模块负责图书资源B模块负责搜索此时A模块新增了图书资源B模块需要同步索引库那同步索引库要如何操作直接放在A模块的业务逻辑中去实现这不符合微服务的理念A模块既要管理数据库又要管理索引库显然是不行的。那通过feign去调用呢显然这样可以实现业务逻辑但系统之间的耦合度又上来了如果后面要扩展一个C模块同样要在A模块新增的时候执行业务那A模块就得改代码。

而使用了RabbitMQ之后可以将消息写入消息队列需要消息的系统自己从消息队列中订阅即可同样是上面那个场景A新增的同时将新增信息存入消息队列B模块需要这个消息从消息队列直接订阅就算后面扩展了C模块同样也从消息队列中订阅即可A模块无需改动任何代码。

可以参考一下图片

  

2.异步

将消息写入消息队列非必要的业务逻辑以异步方式运行加快响应速度。

如假设存在一下的业务执行流程用户下订单成功后发送短信通知发送邮件通知并在app推送通知。这个过程中用户执行完下订单操作后订单服务耗时50ms然后订单服务以异步的方式将消息存入消息队列然后返回给用户响应信息短信服务邮件服务和app推送服务短信服务邮件服务app推送服务耗时都是50ms这样整个业务执行的耗时就是50ms而如果使用传统方式去执行耗时200ms。

3.削峰 

削峰是解决了并发问题在传统模式下并发量太大的时候所有的请求直接怼到数据库造成数据库连接异常。加入消息队列后系统可以根据数据库能处理的并发量分批次从数据库拉取数据。

RabbitMQ的五种消息模型

提一下RabbitMQ官方提供的有6中消息模型只是第六种属于RPC并不是MQ所以这里没有将其列入

Simple-简单模型

在这个模型下RabbitMQ是一个消息代理它接受和转发消息。可以把他想象成一个邮政信箱。

RabbitMQ与邮局的主要区别是它不处理纸张而是接受存储和转发数据消息的二进制数据块。

 

Pproducer/ publisher生产者一个发送消息的用户应用程序。

Cconsumer消费者消费和接收有类似的意思消费者是一个主要用来等待接收消息的用户应用程序

队列红色区域rabbitmq内部类似于邮箱的一个概念。虽然消息流经rabbitmq和你的应用程序但是它们只能存储在队列中。队列只受主机的内存和磁盘限制实质上是一个大的消息缓冲区。许多生产者可以发送消息到一个队列许多消费者可以尝试从一个队列接收数据。

总之

生产者将消息发送到队列消费者从队列中获取消息队列是存储消息的缓冲区。

在这个模式下一个生产者对应一个消费者

代码实现如下

生产者

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.alex_hh.simple.util.ConnectionUtil;

public class Send {

    private final static String QUEUE_NAME = "simple_queue";

    public static void main(String[] argv) throws Exception {
        // 获取到连接以及mq通道
        Connection connection = ConnectionUtil.getConnection();
        // 轻量级的 Connection这是完成大部分API的地方。
        Channel channel = connection.createChannel();

        // 声明创建队列必须声明队列才能够发送消息我们可以把消息发送到队列中。
        // 声明一个队列是幂等的 - 只有当它不存在时才会被创建
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);

        // 消息内容
        String message = "Hello World!";
        channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
        System.out.println(" [x] Sent '" + message + "'");

        //关闭通道和连接
        channel.close();
        connection.close();
    }
}

消费者

import java.io.IOException;
import com.rabbitmq.client.AMQP.BasicProperties;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.Envelope;
import com.alex_hh.simple.util.ConnectionUtil;

public class Recv {
    private final static String QUEUE_NAME = "simple_queue";

    public static void main(String[] argv) throws Exception {
        // 获取到连接
        Connection connection = ConnectionUtil.getConnection();
        // 创建通道
        Channel channel = connection.createChannel();
        // 声明队列
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        // 定义队列的消费者
        DefaultConsumer consumer = new DefaultConsumer(channel) {
            // 获取消息并且处理这个方法类似事件监听如果有消息的时候会被自动调用
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, 
                          BasicProperties properties,byte[] body) throws IOException {
                // body 即消息体
                String msg = new String(body);
                System.out.println(" [消费者1] received : " + msg + "!");
            }
        };
        // 监听队列第二个参数是否自动进行消息确认。
        channel.basicConsume(QUEUE_NAME, true, consumer);
    }
}

消息确定机制ACK

消息一旦被消费者接收队列中的消息就会被删除。

那么问题来了RabbitMQ怎么知道消息被接收了呢

如果消费者领取消息后还没执行操作就挂掉了呢或者抛出了异常消息消费失败但是RabbitMQ无从得知这样消息就丢失了

因此RabbitMQ有一个ACK机制。当消费者获取消息后会向RabbitMQ发送回执ACK告知消息已经被接收。不过这种回执ACK分两种情况

• 自动ACK消息一旦被接收消费者自动发送ACK

• 手动ACK消息接收后不会发送ACK需要手动调用

什么使用自动ACK什么时候使用手动ACK

这需要看消息的重要性

• 如果消息不太重要丢失也没有影响那么自动ACK会比较方便

• 如果消息非常重要不容丢失。那么最好在消费完成后手动ACK否则接收消息后就自动ACKRabbitMQ就会把消息从队列中删除。如果此时消费者宕机那么消息就丢失了。

开启手动ACK

 修改消费者代码

public static void main(String[] argv) throws Exception {
        // 获取到连接
        Connection connection = ConnectionUtil.getConnection();
        // 创建通道
        Channel channel = connection.createChannel();
        // 声明队列
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        // 定义队列的消费者
        DefaultConsumer consumer = new DefaultConsumer(channel) {
            // 获取消息并且处理这个方法类似事件监听如果有消息的时候会被自动调用
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, 
                          BasicProperties properties,byte[] body) throws IOException {
                // body 即消息体
                String msg = new String(body);
                System.out.println(" [消费者1] received : " + msg + "!");
                //开启手动ACK
                channel.basicAck(envelope.getDeliveryTag(),false);
            }
        };
        // 监听队列第二个参数是否自动进行消息确认。
        channel.basicConsume(QUEUE_NAME, false, consumer);
    }

Work-工作模型

 工作队列又称任务队列。主要思想就是避免执行资源密集型任务时必须等待它执行完成。相反我们稍后完成任务我们将任务封装为消息并将其发送到队列。 在后台运行的工作进程将获取任务并最终执行作业。当你运行许多消费者时任务将在他们之间共享但是一个消息只能被一个消费者获取。

这个概念在Web应用程序中特别有用因为在短的HTTP请求窗口中无法处理复杂的任务。

在上面的基础上做些改进调整生产者生产多条消息

public class Send {
    private final static String QUEUE_NAME = "test_work_queue";

    public static void main(String[] argv) throws Exception {
        // 获取到连接
        Connection connection = ConnectionUtil.getConnection();
        // 获取通道
        Channel channel = connection.createChannel();
        // 声明队列
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        // 循环发布任务
        for (int i = 0; i < 50; i++) {
            // 消息内容
            String message = "task .. " + i;
            channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
            System.out.println(" [x] Sent '" + message + "'");

            Thread.sleep(i * 2);
        }
        // 关闭通道和连接
        channel.close();
        connection.close();
    }
}

创建一个新的消费者2

public static void main(String[] argv) throws Exception {
        // 获取到连接
        Connection connection = ConnectionUtil.getConnection();
        // 创建通道
        Channel channel = connection.createChannel();
        // 声明队列
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        // 定义队列的消费者
        DefaultConsumer consumer = new DefaultConsumer(channel) {
            // 获取消息并且处理这个方法类似事件监听如果有消息的时候会被自动调用
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, 
                          BasicProperties properties,byte[] body) throws IOException {
                // body 即消息体
                String msg = new String(body);
                System.out.println(" [消费者2] received : " + msg + "!");
                try{
                    //模拟任务完成耗时
                    Thread.sleep(1000);
                }catch(Exception e){
                    
                }
                //开启手动ACK
                channel.basicAck(envelope.getDeliveryTag(),false);
            }
        };
        // 监听队列第二个参数是否自动进行消息确认。
        channel.basicConsume(QUEUE_NAME, false, consumer);
    }

两个消费者一同启动然后生产者发送50条消息可以发现两个消费者各自消费了25条消息而且各不相同这就实现了任务的分发。

 

但是其中存在问题

• 消费者1比消费者2的效率要低一次任务的耗时较长

• 然而两人最终消费的消息数量是一样的

• 消费者2大量时间处于空闲状态消费者1一直忙碌

现在的状态属于是把任务平均分配正确的做法应该是消费越快的人消费的越多。

怎么实现呢

我们可以使用basicQos方法和prefetchCount = 1设置。 这告诉RabbitMQ一次不要向工作人员发送多于一条消息。 或者换句话说不要向工作人员发送新消息直到它处理并确认了前一个消息。 相反它会将其分派给不是仍然忙碌的下一个工作人员。

在消费者上加入

channel.basicQos(1);

 再次执行

 Fanout-广播模型

在广播模式下消息发送流程是这样的

• 1 可以有多个消费者

• 2 每个消费者有自己的queue队列

• 3 每个队列都要绑定到Exchange交换机

• 4 生产者发送的消息只能发送到交换机交换机来决定要发给哪个队列生产者无法决定。

• 5 交换机把消息发送给绑定过的所有队列

• 6 队列的消费者都能拿到消息。实现一条消息被多个消费者消费

 

 在广播模式下生产者不在将消息发送至队列而是发送给交换机由交换机来实现消息的分配改一下生产者的代码

public class Send {

    private final static String EXCHANGE_NAME = "fanout_exchange_test";

    public static void main(String[] argv) throws Exception {
        // 获取到连接
        Connection connection = ConnectionUtil.getConnection();
        // 获取通道
        Channel channel = connection.createChannel();
        
        // 声明exchange指定类型为fanout
        channel.exchangeDeclare(EXCHANGE_NAME, "fanout");
        
        // 消息内容
        String message = "Hello everyone";
        // 发布消息到Exchange
        channel.basicPublish(EXCHANGE_NAME, "", null, message.getBytes());
        System.out.println(" [生产者] Sent '" + message + "'");

        channel.close();
        connection.close();
    }
}

消费者需要声明队列并绑定交换机更改代码如下

消费者1

public class Recv {
    private final static String QUEUE_NAME = "fanout_exchange_queue_1";

    private final static String EXCHANGE_NAME = "fanout_exchange_test";

    public static void main(String[] argv) throws Exception {
        // 获取到连接
        Connection connection = ConnectionUtil.getConnection();
        // 获取通道
        Channel channel = connection.createChannel();
        // 声明队列
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);

        // 绑定队列到交换机
        channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "");

        // 定义队列的消费者
        DefaultConsumer consumer = new DefaultConsumer(channel) {
            // 获取消息并且处理这个方法类似事件监听如果有消息的时候会被自动调用
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, 
                          BasicProperties properties,byte[] body) throws IOException {
                // body 即消息体
                String msg = new String(body);
                System.out.println(" [消费者1] received : " + msg + "!");
            }
        };
        // 监听队列自动返回完成
        channel.basicConsume(QUEUE_NAME, true, consumer);
    }
}

消费者2

public class Recv2 {
    private final static String QUEUE_NAME = "fanout_exchange_queue_2";

    private final static String EXCHANGE_NAME = "fanout_exchange_test";

    public static void main(String[] argv) throws Exception {
        // 获取到连接
        Connection connection = ConnectionUtil.getConnection();
        // 获取通道
        Channel channel = connection.createChannel();
        // 声明队列
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);

        // 绑定队列到交换机
        channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "");
        
        // 定义队列的消费者
        DefaultConsumer consumer = new DefaultConsumer(channel) {
            // 获取消息并且处理这个方法类似事件监听如果有消息的时候会被自动调用
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, 
                          BasicProperties properties,byte[] body) throws IOException {
                // body 即消息体
                String msg = new String(body);
                System.out.println(" [消费者2] received : " + msg + "!");
            }
        };
         // 监听队列自动返回完成
        channel.basicConsume(QUEUE_NAME, true, consumer);
    }
}

 测试运行两个消费者并启动生产者发送一条消息

Direct-定向模型

有选择性的接收消息

在订阅模式广播中生产者发布消息所有消费者都可以获取所有消息。

在路由模式定向中我们将添加一个功能 - 我们将只能订阅一部分消息。 例如我们只能将重要的错误消息引导到日志文件以节省磁盘空间同时仍然能够在控制台上打印所有日志消息。

但是在某些场景下我们希望不同的消息被不同的队列消费。这时就要用到Direct类型的Exchange。

在Direct模型下队列与交换机的绑定不能是任意绑定了而是要指定一个RoutingKey路由key

消息的发送方在向Exchange发送消息时也必须指定消息的routing key。

 

P生产者向Exchange发送消息发送消息时会指定一个routing key。

XExchange交换机接收生产者的消息然后把消息递交给 与routing key完全匹配的队列

C1消费者其所在队列指定了需要routing key 为 error 的消息

C2消费者其所在队列指定了需要routing key 为 info、error、warning 的消息

 与前面基本相似在生产者声明交换机时指定类型为“direct”

channel.exchangeDeclare(EXCHANGE_NAME, "direct");

 发送消息时指定routing key

channel.basicPublish(EXCHANGE_NAME, "insert", null, message.getBytes());

 消费者绑定交换机时指定routingkey

channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "insert");

 Topic-主题模型

Topic类型的Exchange与Direct相比都是可以根据RoutingKey把消息路由到不同的队列。只不过Topic类型Exchange可以让队列在绑定Routing key 的时候使用通配符

Routingkey 一般都是有一个或多个单词组成多个单词之间以”.”分割例如 item.insert

通配符规则

#匹配一个或多个词

*匹配不多不少恰好1个词

如

audit.#能够匹配audit.irs.corporate 或者 audit.irs

audit.*只能匹配audit.irs

 使用与Direct模式基本一样只是生产者在声明交换机的时候指定类型为‘topic’

channel.exchangeDeclare(EXCHANGE_NAME, "topic"); 

 消息的持久化

如何避免消息丢失

1 消费者的手动ACK机制。可以防止业务处理失败。

2 但是如果在消费者消费之前MQ就宕机了消息就没了。

是可以将消息进行持久化呢

要将消息持久化前提是队列、Exchange都持久化

交换机的持久化

//生产者声明交换机的时候在第三个属性durable配置为true即可

 channel.exchangeDeclare(EXCHANGE_NAME, "topic",true); 

队列的持久化

消费者声明队列的时候将第二个属性durable配置为true即可

channel.queueDeclare(QUEUE_NAME, true, false, false, null);

消息的持久化

生产者发送消息时指定第三个参数为

channel.basicPublish(EXCHANGE_NAME, "insert", MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes());

 总结一下五种模型simple和work都是生产者直接发送消息到消息队列而simple是一对一的关系而work是一对多的关系

fanout,direct,topic都是通过交换机来发送消息不同在于fanout发送的消息所有订阅到交换机上的消费者都能获取direct与topic可以指定routing key来让部分持有routing key的订阅者获取消息而topic可以指定通配符direct不行。

 如何实现可靠消息最终一致性

在实际系统的开发过程中可能服务间的调用是异步的。也就是说一个服务发送一个消息给 MQ即消息中间件比如RocketMQ、RabbitMQ、Kafka、ActiveMQ 等等。

然后另外一个服务从 MQ 消费到一条消息后进行处理。这就成了基于 MQ 的异步调用了。

那么针对这种基于 MQ 的异步调用如何保证各个服务间的分布式事务呢也就是说我希望的是基于MQ 实现异步调用的多个服务的业务逻辑要么一起成功要么一起失败。这个时候就要用上可靠消息最终一致性方案来实现分布式事务。

可靠消息最终一致性方案是指当事务发起方执行完成本地事务后并发出一条消息事务参与方(消息消费者)一定能够接收消息并处理事务成功此方案强调的是只要消息发给事务参与方最终事务要达到一致。

此方案是利用消息中间件完成如下图

事务发起方消息生产方将消息发给消息中间件事务参与方从消息中间件接收消息事务发起方和消息中间件之间事务参与方消息消费方和消息中间件之间都是通过网络通信由于网络通信的不确定性会导致分布式事务问题。

可靠消息一致性需要解决的问题

1.上游服务把消息成功发送

本地事务与消息发送的原子性问题事务发起方在本地事务执行成功后消息必须发出去否则就回滚事务。即实现本地事务和消息发送的原子性要么都成功要么都失败。可靠消息

2.下游服务把消息成功消费

事务参与方接收消息的可靠性事务参与方必须能够从消息队列接收到消息。可靠消息

3.对消息做幂等处理

消息重复消费的问题由于网络2的存在若某一个消费节点响应超时但是消费成功此时消息中间件会重复投递此消息就导致了消息的重复消费。最终一致性

解决方案

为了让上游服务把消息成功发出可以使用本地消息表该方案最初是eBay提出的在系统A处理任务完成后在本地记录待发送信息。一个定时任务不断检查是否发送成功如果发送成功将记录状态修改。如下图

 如处理任务后在本地消息表中添加一条数据发送消息到中间件随后通过异步的方式等待响应如果响应成功返回则修改数据的记录状态也可以直接删除如果收不到响应则通过定时任务一直发送直至成功。

为了让下游服务把消息成功消费可以使用

消息持久化可保证消息中间件宕机后消息不丢失

手动ack保证消息投递失败时消息的重新投递

 实现消息的幂等性可以通过消息去重表。

消息去重表任务B处理消息前先查询该消息是否被消费如果没消费处理任务B成功记录消息。如果消息已经被消费直接返回应答成功

 代码实现

创建本地消息记录表

DROP TABLE IF EXISTS `local_message`;
CREATE TABLE `local_message` (
  `tx_no` varchar(255) NOT NULL,
  `item_id` bigint DEFAULT NULL,
  `state` int(11) DEFAULT NULL,
  PRIMARY KEY (`tx_no`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8;

新增信息去重表

DROP TABLE IF EXISTS `msg_distinct`;
CREATE TABLE `msg_distinct` (
  `tx_no` varchar(255) NOT NULL,
  `create_time` datetime DEFAULT NULL,
  PRIMARY KEY (`tx_no`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8; 

 服务提供者工程导入quartz依赖

        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-quartz</artifactId>
        </dependency>

服务提供者yml:

rabbitmq: #rabbitmq配置
  host: 192.168.40.146
  port: 5672
  username: admin
  password: 1111
  virtual-host: /
  publisher-returns: true #开启消息退回回调
  publisher-confirm-type: correlated #开启消息确认回调
  listener:
    direct:
      acknowledge-mode: manual #开启交换机模式手动ack
    simple:
      acknowledge-mode: manual #开启直连模式手动ack

 service:

    @Override
    @Transactional(rollbackFor = {Exception.class})
    public void insertTbItem(TbItemVo tbItem) {
        //修改需要创建修改时间并且同步修改在关联表tb_item_param_item中
        TbItem param = new TbItem();
        //将工具实体类的值赋值给item对象
        BeanUtils.copyProperties(tbItem,param);
        //设置创建时间
        param.setCreated(new Date());
        param.setUpdated(new Date());
        //设置当前状态
        param.setStatus((byte) 1);
        //设置id
        long id = IDUtils.genItemId();
        param.setId(id);
        //调用mapper修改数据
        tbItemMapper.insert(param);
        //同时修改关联数据
        TbItemParamItem paramItem = new TbItemParamItem();
        //设置参数
        paramItem.setParamData(tbItem.getItemParams());
        //设置关联id
        paramItem.setItemId(id);
        //设置时间
        paramItem.setCreated(new Date());
        paramItem.setUpdated(new Date());
        tbItemParamItemMapper.insert(paramItem);
        //设置关联item_desc
        TbItemDesc desc = new TbItemDesc();
        desc.setItemDesc(tbItem.getDesc());
        desc.setItemId(id);
        desc.setCreated(new Date());
        desc.setUpdated(new Date());
        tbItemDescMapper.insert(desc);
        //保存新增信息至本地消息表
        LocalMessage localMessage = new LocalMessage();
        localMessage.setTxNo(UuidUtils.generateUuid().toString());
        localMessage.setItemId(id);
        localMessage.setState(0);
        localMessageMapper.insertSelective(localMessage);
    }

 这里service执行玩新增后调用持久层在本地信息记录表中记录新增数据

编写消息发送类

/**
 * 消息发送者
 */
@Component
public class RabbitMQSender implements ConfirmCallback, ReturnCallback {

    private final Logger logger = LoggerFactory.getLogger(RabbitMQSender.class);

    @Autowired
    private LocalMessageMapper localMessageMapper;

    @Autowired
    private AmqpTemplate amqpTemplate;

    private String exchange = String.valueOf(RabbitMQEnum.ITEM_EXCHANGE);

    private String routingKey = String.valueOf(RoutingKey.ITEM_INSERT);

    /**
     * 发送信息至交换机
     * @param localMessage
     */
    public void sendMsg(LocalMessage localMessage){
        RabbitTemplate rabbitTemplate = (RabbitTemplate) amqpTemplate;
        //设置确认回调和失败回调
        rabbitTemplate.setConfirmCallback(this);
        rabbitTemplate.setReturnCallback(this);
        //创建相关消息对象
        CorrelationData correlationData = new CorrelationData(localMessage.getTxNo());
        //发送信息需要指定交换机路由id,发送的信息相关信息
        rabbitTemplate.convertAndSend(exchange,routingKey, JsonUtils.objectToJson(localMessage));
    }

    /**
     * 确认回调方法
     * @param correlationData
     * @param ack
     * @param cause
     */
    @Override
    public void confirm(CorrelationData correlationData, boolean ack, String cause) {
        if(ack){
            //消息发送成功更新本地信息为已发送成功状态或者直接删除本地记录
            String txNo = correlationData.getId();
            localMessageMapper.deleteByPrimaryKey(txNo);
        }
    }

    /**
     * 失败回调方法
     * @param message
     * @param replyCode
     * @param replyText
     * @param exchange
     * @param routingKey
     */
    @Override
    public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
            logger.info("return--message:" + new String(message.getBody())
                    + ",exchange:" + exchange + ",routingKey:" + routingKey);
    }
}

如果消息发送成功则会回调确认回调方法删除本地信息记录表的数据

失败则调用失败回调方法打印日志。

编写定时任务类

/**
 * 任务类定时发送信息
 */
@Component
public class ItemQuartz {

    private Logger logger = LoggerFactory.getLogger(ItemQuartz.class);

    @Autowired
    private LocalMessageMapper localMessageMapper;

    @Autowired
    private RabbitMQSender rabbitMQSender;

    /**
     * 定时任务
     */
    public void scanLocalMessage(){
        logger.info("执行扫描本地消息表的任务" + new Date());
        List<LocalMessage> localMessages = localMessageMapper.selectByExample(null);
        if(!CollectionUtils.isEmpty(localMessages)){
            for (LocalMessage localMessage : localMessages) {
                rabbitMQSender.sendMsg(localMessage);
            }
        }

    }
}

 定时任务执行扫描操作如果本地信息表中存在消息则将其发送。

定时任务配置

/**
 * 定时任务配置
 */
@Configuration
public class QuartzConfig {
    //定义工作任务
    @Bean
    public MethodInvokingJobDetailFactoryBean methodInvokingJobDetailFactoryBean(ItemQuartz itemQuartz){
        MethodInvokingJobDetailFactoryBean jobDetailFactoryBean =
                new MethodInvokingJobDetailFactoryBean();
        jobDetailFactoryBean.setTargetObject(itemQuartz);
        jobDetailFactoryBean.setTargetMethod("scanLocalMessage");
        return jobDetailFactoryBean;
    }
    
    //定义触发器
    @Bean
    public CronTriggerFactoryBean cronTriggerFactoryBean(MethodInvokingJobDetailFactoryBean jobDetailFactoryBean){
        CronTriggerFactoryBean triggerFactoryBean = new CronTriggerFactoryBean();
        triggerFactoryBean.setCronExpression("*/1 * * * * ?");
        triggerFactoryBean.setJobDetail(jobDetailFactoryBean.getObject());
        return triggerFactoryBean;
    }

    //scheduled什么时候做什么事
    @Bean
    public SchedulerFactoryBean schedulerFactoryBean(
            CronTriggerFactoryBean triggerFactoryBean) {
        SchedulerFactoryBean schedulerFactoryBean = new SchedulerFactoryBean();
        schedulerFactoryBean.setTriggers(triggerFactoryBean.getObject());
        return schedulerFactoryBean;
    }
}

配置定时任务的执行。

如此便解决了 本地事务与消息发送的原子性问题以及消息数据持久化的问题convertAndSent方法发送消息默认持久化

服务消费者yml配置

spring:
    listener:
      direct:
        acknowledge-mode: manual #手动确认
      simple:
        acknowledge-mode: manual #手动确认

服务消费者从消息队列中获取消息

/**
 * 监听消息队列
 */
@Component
public class SearchItemListener {

    private Logger logger = LoggerFactory.getLogger(SearchItemListener.class);

    @Autowired
    private SearchItemService searchItemService;

    @Autowired
    private MsgDistinctService msgDistinctService;

    /**
     * 监听mq队列中的消息
     * @param msg 接收到的消息
     * @param channel 信道
     * @param message 封装的信息对象
     */
    @RabbitListener(bindings = {
            @QueueBinding(
                    value = @Queue(value = "item_queue",durable = "true"),
                    exchange = @Exchange(value = "item_exchange",type = ExchangeTypes.TOPIC),
                    key = {"item.*"}
            )
    })
    public void listen(String  msg, Channel channel, Message message) throws IOException {
        logger.info("接收到消息" + msg);
        LocalMessage localMessage = JsonUtils.jsonToPojo(msg, LocalMessage.class);
        //进行幂等判断
        MsgDistinct msgDistinct =msgDistinctService.selectMsgDistinctByTxNo(localMessage.getTxNo());
        if(msgDistinct == null){
            searchItemService.addDoc(localMessage.getItemId());
            msgDistinctService.insertMsgDistinct(localMessage.getTxNo());
        }else{
            System.out.println("=======幂等生效事务"+msgDistinct.getTxNo()
                    +" 已成功执行===========");
        }
        channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);
    }
}

 通过注解的方式配置交换机默认持久与队列持久化通过消息去重表保证幂等性通过手动ack保证消息投递失败时的重新投递当表里存在本地消息记录表的数据时直接提交不存在则添加记录并提交。如此便保证了可靠消息最终一致性