Java RabbitMQ advanced feature example analysis

Reliable delivery of messages

When using RabbitMQ, as the message sender, we hope to prevent any message loss or delivery failure scenarios. RabbitMQ provides us with two ways to control the delivery reliability mode of messages.

• confirm confirmation mode

• return return mode

rabbitmq’s entire message delivery path is:

producer—>rabbitmq broker—>exchange—>queue—>consumer

• ##The message from producer to exchange will return a confirmCallback

• If the message fails to be delivered from the exchange—>queue, a returnCallback will be returned

##We can use these two callbacks to control the reliability of message delivery

Confirmation mode

When the message is sent from producer to exchange, a confirmCallback will be returned

Take spring integration rabbitmq as an example, modify rabbitmq configuration file, add publisher-confirms attribute in connectionFactory and set the value Is true

<!--
* 确认模式：
* 步骤：
* 1. 确认模式开启：ConnectionFactory中开启publisher-confirms="true"
-->
<!-- 定义rabbitmq connectionFactory -->
    <rabbit:connection-factory id="connectionFactory" host="${rabbitmq.host}"
                               port="${rabbitmq.port}"
                               username="${rabbitmq.username}"
                               password="${rabbitmq.password}"
                               virtual-host="${rabbitmq.virtual-host}"
                               publisher-confirms="true"/>

/*
 * 确认模式：
 * 步骤：
 * 2. 在rabbitTemplate定义ConfirmCallBack回调函数
 */
@Test
    public void queueTest(){
        rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {
            @Override
            public void confirm(CorrelationData correlationData, boolean ack, String cause) {
            /**
             *
             * @param correlationData 相关配置信息
             * @param ack exchange交换机 是否成功收到了消息。true 成功，false代表失败
             * @param cause 失败原因
             */
                System.out.println("confirm方法被执行了....");
                if (ack) {
                    //接收成功
                    System.out.println("接收成功消息" + cause);
                } else {
                    //接收失败
                    System.out.println("接收失败消息" + cause);
                    //做一些处理，让消息再次发送。
                }
            }
        });
        //路由键与队列同名
        rabbitTemplate.convertAndSend("spring_queue", "message confirm....");
    }

Because the message is sent to the queue normally, the returned cause value is empty. If an exception occurs, cause is the abnormal reason

Return Mode

If the message delivery from the exchange–>queue fails, a returnCallback will be returned

1. Turn on the fallback mode: publisher-returns="true"

    <!-- 定义rabbitmq connectionFactory -->
    <rabbit:connection-factory id="connectionFactory" host="${rabbitmq.host}"
                               port="${rabbitmq.port}"
                               username="${rabbitmq.username}"
                               password="${rabbitmq.password}"
                               virtual-host="${rabbitmq.virtual-host}"
                               publisher-returns="true"/>

2. Set the mode for Exchange message processing failure: setMandatory, and then set ReturnCallBack

    @Test
    public void queueTest(){
        //1.设置交换机处理失败消息的模式
        rabbitTemplate.setMandatory(true);
        //2.设置ReturnCallBack
        rabbitTemplate.setReturnCallback(new RabbitTemplate.ReturnCallback() {
            /**
             * @param message    消息对象
             * @param replyCode  错误码
             * @param replyText  错误信息
             * @param exchange   交换机
             * @param routingKey 路由键
             */
            @Override
            public void returnedMessage(Message message, int replyCode, String
                    replyText, String exchange, String routingKey) {
                System.out.println("return 执行了....");
                System.out.println(message);
                System.out.println(replyCode);
                System.out.println(replyText);
                System.out.println(exchange);
                System.out.println(routingKey);
                //处理
            }
        });
        //手动添加错误路由模拟错误发生
        rabbitTemplate.convertAndSend("spring_topic_exchange", "return123", "return message...");
    }

Only when an error occurs will the message be returned, so manually add an error and add the routing value return123 to the sent message. In fact, there is no such Routing, the message returned by running is as follows.

Consumer Ack

ack refers to Acknowledge, confirmation. Indicates the confirmation method after the consumer receives the message.

There are three confirmation methods:

Automatic confirmation: acknowledge="none"
Manual confirmation: acknowledge="manual ”
Confirm according to abnormal conditions: acknowledge="auto", (This method is troublesome to use and does not require learning)
Among them, automatic confirmation It means that once the message is received by the Consumer, the receipt will be automatically confirmed and the corresponding message will be removed from the RabbitMQ message cache. However, in actual business processing, it is very likely that the message will be lost if there is an exception in the business processing after the message is received. If the manual confirmation method is set, you need to call channel.basicAck() after the business processing is successful and sign in manually. If an exception occurs, call the channel.basicNack() method to let it automatically resend the message.

Let's take spring's integration of rabbitmq as an example. Set the confirmation method in the rabbitmq configuration file.

<rabbit:listener-container connection-factory="connectionFactory"
acknowledge="manual">
.....

The monitoring class code is as follows:

public class AckListener implements ChannelAwareMessageListener {
    @Override
    public void onMessage(Message message, Channel channel) throws Exception {
        long deliveryTag = message.getMessageProperties().getDeliveryTag();
        try {
            //1.接收转换消息
            System.out.println(new String(message.getBody()));
            //2. 处理业务逻辑
            System.out.println("处理业务逻辑...");
            int i = 3/0;//出现错误
            // 3. 手动签收
            channel.basicAck(deliveryTag,true);
        } catch (Exception e) {
            //e.printStackTrace();
            //4.拒绝签收
            /*
             *第三个参数：requeue：重回队列。如果设置为true，则消息重新回到queue，broker会
             *重新发送该消息给消费端
             */
            channel.basicNack(deliveryTag,true,true);
            //channel.basicReject(deliveryTag,true);
        }
    }
}

The channel.basicNack() method is called because of an exception. , let it automatically resend the message, so the output content is infinite loop

Consumer side current limit

When our When the Rabbitmq server has a backlog of tens of thousands of unprocessed messages, if we open a consumer client at will, the following situation will occur: a huge amount of messages are pushed over instantly, but our single client cannot process so much data at the same time! When When the amount of data is particularly large, it is definitely unscientific for us to limit the flow of the production end, because sometimes the concurrency is extremely large, and sometimes the concurrency is extremely small, and we cannot restrict the production end. This is the user's behavior. Therefore, we should limit the flow on the consumer side. rabbitmq provides a qos (quality of service) function, that is, on the premise of non-automatic confirmation of messages, if a certain number of messages (Qos value is set for channel or consumer) are not confirmed before , do not consume new messages.

1. Ensure that the ack mechanism is manual confirmation

2. The listener-container configuration attribute perfetch = 1 means that the consumer pulls a message from mq for consumption each time until the manual confirmation of consumption is completed. After that, it will continue to pull the next message.

<rabbit:listener-container connection-factory="connectionFactory" auto-declare="true" acknowledge="manual" prefetch="1">
        <rabbit:listener ref="topicListenerACK" queue-names="spring_topic_queue_well2"/>
</rabbit:listener-container>

Producer, send five messages

    @Test
    public void topicTest(){
/**
 * 参数1：交换机名称
 * 参数2：路由键名
 * 参数3：发送的消息内容
 */
        for (int i=0;i<5;i++){
            rabbitTemplate.convertAndSend("spring_topic_exchange", "xzk.a", "发送到spring_topic_exchange交换机xzk.cn的消息"+i);
        }
    }
}

The producer comments out channel.basicAck(deliveryTag,true) and does not confirm receipt of the message

public class AckListener implements ChannelAwareMessageListener {
    @Override
    public void onMessage(Message message, Channel channel) throws Exception {
        long deliveryTag = message.getMessageProperties().getDeliveryTag();
        try {
            //1.接收转换消息
            System.out.println(new String(message.getBody()));
            //2. 处理业务逻辑
            System.out.println("处理业务逻辑...");
            // 3. 手动签收
            //channel.basicAck(deliveryTag,true);
        } catch (Exception e) {
            //e.printStackTrace();
            //4.拒绝签收
            /*
             *第三个参数：requeue：重回队列。如果设置为true，则消息重新回到queue，broker会
             *重新发送该消息给消费端
             */
            channel.basicNack(deliveryTag,true,true);
        }
    }
}

Start the consumer at this time After running the producer again, I found that the consumer sent five messages. In fact, the producer only received one message, which reached the current limiting effect.

Observed the rabbitmq console and found that there are 1 unack message. 4 ready messages have not reached the consumer yet. It is consistent with the current limiting situation of prefetchCount=1 we set.

把channel.basicAck(deliveryTag,true)的注释取消掉，即可以自动确认收到消息，重新运行消费者，接收到了另外的四条消息

TTL(Time To Live)

Time To Live，消息过期时间设置

设置某个队列为过期队列

设置交换机，队列以及队列过期时间为10000ms

 <!--ttl-->
    <rabbit:queue name="test_queue_ttl" id="test_queue_ttl">
        <rabbit:queue-arguments>
            <entry key="x-message-ttl" value="10000" value-type="java.lang.Integer"/>
        </rabbit:queue-arguments>
    </rabbit:queue>
    <rabbit:topic-exchange name="test_exchange_ttl">
        <rabbit:bindings>
            <rabbit:binding pattern="ttl.#" queue="test_queue_ttl"/>
        </rabbit:bindings>
    </rabbit:topic-exchange>

生产者发送10条消息

    @Test
    public void testTtl() {
        for (int i = 0; i < 10; i++) {
            rabbitTemplate.convertAndSend("test_exchange_ttl","ttl.hehe","message ttl...");
        }

十秒钟后，过期消息消失

设置单独某个消息过期

设置交换机和队列

<rabbit:queue name="test_queue_ttl" id="test_queue_ttl"/>
<rabbit:topic-exchange name="test_exchange_ttl">
    <rabbit:bindings>
        <rabbit:binding pattern="ttl.#" queue="test_queue_ttl"/>     
    </rabbit:bindings>
</rabbit:topic-exchange>

生产者发送特定过期消息，用到了MessagePostProcessor这个api

 @Test
    public void testTtl() {
        MessagePostProcessor messagePostProcessor = new MessagePostProcessor() {
            @Override
            public Message postProcessMessage(Message message) throws AmqpException {
                //1.设置message信息
                message.getMessageProperties().setExpiration("5000");//消息的过期时间
                //2.返回该消息
                return message;
            }
        };
        //消息单独过期
        rabbitTemplate.convertAndSend("test_exchange_ttl","ttl.hehe","message ttl...",messagePostProcessor);
    }

5s之后

注:

1.如果同时设置队列过期和消息过期，系统会根据哪个过期的时间短而选用哪儿个。

2.设置单独消息过期时，如果该消息不为第一个接受的消息，则不过期。

死信队列

死信队列，英文缩写：DLX 。Dead Letter Exchange（死信交换机），当消息成为Deadmessage后，可以被重新发送到另一个交换机，这个交换机就是DLX。

消息成为死信的三种情况：

• 队列消息长度到达限制；

• 消费者拒接消费消息，basicNack/basicReject,并且不把消息重新放入原目标队列,requeue=false；

• 原队列存在消息过期设置，消息到达超时时间未被消费；

队列绑定死信交换机：

给队列设置参数： x-dead-letter-exchange 和 x-dead-letter-routing-key

实现

1.声明正常的队列(test_queue_dlx)和交换机(test_exchange_dlx)

<rabbit:queue name="test_queue_dlx" id="test_queue_dlx">
    <!--正常队列绑定死信交换机-->
    <rabbit:queue-arguments>
        <!--x-dead-letter-exchange：死信交换机名称-->
        <entry key="x-dead-letter-exchange" value="exchange_dlx" />
        <!--3.2 x-dead-letter-routing-key：发送给死信交换机的routingkey-->
        <entry key="x-dead-letter-routing-key" value="dlx.hehe" />
        <!--4.1 设置队列的过期时间 ttl-->
        <entry key="x-message-ttl" value="10000" value-type="java.lang.Integer"/>
        <!--4.2 设置队列的长度限制 max-length -->
        <entry key="x-max-length" value="10" value-type="java.lang.Integer" />
    </rabbit:queue-arguments>
</rabbit:queue>
<rabbit:topic-exchange name="test_exchange_dlx">
    <rabbit:bindings>
        <rabbit:binding pattern="test.dlx.#" queue="test_queue_dlx">
        </rabbit:binding>
    </rabbit:bindings>
</rabbit:topic-exchange>

2.声明死信队列(queue_dlx)和死信交换机(exchange_dlx)

<rabbit:queue name="queue_dlx" id="queue_dlx"></rabbit:queue>
<rabbit:topic-exchange name="exchange_dlx">
    <rabbit:bindings>
        <rabbit:binding pattern="dlx.#" queue="queue_dlx"></rabbit:binding>
    </rabbit:bindings>
</rabbit:topic-exchange>

3.生产端测试

/**
* 发送测试死信消息：
* 1. 过期时间
* 2. 长度限制
* 3. 消息拒收
*/
@Test
public void testDlx(){
    //1. 测试过期时间，死信消息
    rabbitTemplate.convertAndSend("test_exchange_dlx","test.dlx.haha","我是一条消息，我会死吗？");
    //2. 测试长度限制后，消息死信
    /* for (int i = 0; i < 20; i++) {
    rabbitTemplate.convertAndSend("test_exchange_dlx","test.dlx.haha","我是一条消息，我会死吗？");
    }*/
    //3. 测试消息拒收
    //rabbitTemplate.convertAndSend("test_exchange_dlx","test.dlx.haha","我是一条消息，我会死吗？");
}

4.消费端监听

public class DlxListener implements ChannelAwareMessageListener {
    @Override
    public void onMessage(Message message, Channel channel) throws Exception {
        long deliveryTag = message.getMessageProperties().getDeliveryTag();
        try {
            //1.接收转换消息
            System.out.println(new String(message.getBody()));
            //2. 处理业务逻辑
            System.out.println("处理业务逻辑...");
            int i = 3/0;//出现错误
            //3. 手动签收
            channel.basicAck(deliveryTag,true);
        } catch (Exception e) {
            //e.printStackTrace();
            System.out.println("出现异常，拒绝接受");
            //4.拒绝签收，不重回队列 requeue=false
            channel.basicNack(deliveryTag,true,false);
        }
    }
}

<rabbit:listener ref="dlxListener" queue-names="test_queue_dlx">
</rabbit:listener>

延迟队列

延迟队列，即消息进入队列后不会立即被消费，只有到达指定时间后，才会被消费。c

需求：

1.下单后，30分钟未支付，取消订单，回滚库存。

2.新用户注册成功7天后，发送短信问候。

实现方式：

• 定时器

• 延迟队列

定时器的实现方式不够优雅，我们采取延迟队列的方式

不过很可惜，在RabbitMQ中并未提供延迟队列功能。

但是可以使用：TTL+死信队列 组合实现延迟队列的效果。

配置

<!--
延迟队列：
        1. 定义正常交换机（order_exchange）和队列(order_queue)
        2. 定义死信交换机（order_exchange_dlx）和队列(order_queue_dlx)
        3. 绑定，设置正常队列过期时间为30分钟
-->
<!-- 定义正常交换机（order_exchange）和队列(order_queue)-->
<rabbit:queue id="order_queue" name="order_queue">
<!-- 绑定，设置正常队列过期时间为30分钟-->
    <rabbit:queue-arguments>
        <entry key="x-dead-letter-exchange" value="order_exchange_dlx" />
        <entry key="x-dead-letter-routing-key" value="dlx.order.cancel" />
        <entry key="x-message-ttl" value="10000" value-type="java.lang.Integer"/>
    </rabbit:queue-arguments>
</rabbit:queue>
<rabbit:topic-exchange name="order_exchange">
    <rabbit:bindings>
        <rabbit:binding pattern="order.#" queue="order_queue"></rabbit:binding>
    </rabbit:bindings>
</rabbit:topic-exchange>
<!-- 定义死信交换机（order_exchange_dlx）和队列(order_queue_dlx)-->
<rabbit:queue id="order_queue_dlx" name="order_queue_dlx"></rabbit:queue>
<rabbit:topic-exchange name="order_exchange_dlx">
    <rabbit:bindings>
        <rabbit:binding pattern="dlx.order.#" queue="order_queue_dlx"></rabbit:binding>
    </rabbit:bindings>
</rabbit:topic-exchange>

生产端测试

@Test
public void testDelay() throws InterruptedException {
    //1.发送订单消息。 将来是在订单系统中，下单成功后，发送消息
    rabbitTemplate.convertAndSend("order_exchange","order.msg","订单信息：id=1,time=2019年8月17日16:41:47");
    /*//2.打印倒计时10秒
    for (int i = 10; i > 0 ; i--) {
        System.out.println(i+"...");
        Thread.sleep(1000);
    }*/
}

消费端监听

public class OrderListener implements ChannelAwareMessageListener {
    @Override
    public void onMessage(Message message, Channel channel) throws Exception {
		long deliveryTag = message.getMessageProperties().getDeliveryTag();
		try {
			//1.接收转换消息
			System.out.println(new String(message.getBody()));
			//2. 处理业务逻辑
			System.out.println("处理业务逻辑...");
			System.out.println("根据订单id查询其状态...");
			System.out.println("判断状态是否为支付成功");
			System.out.println("取消订单，回滚库存....");
			//3. 手动签收
			channel.basicAck(deliveryTag,true);
		} catch (Exception e) {
			//e.printStackTrace();
			System.out.println("出现异常，拒绝接受");
			//4.拒绝签收，不重回队列 requeue=false
			channel.basicNack(deliveryTag,true,false);
		}
	}
}

<rabbit:listener ref="orderListener" queue-names="order_queue_dlx">
</rabbit:listener>

The above is the detailed content of Java RabbitMQ advanced feature example analysis. For more information, please follow other related articles on the PHP Chinese website!