java——spring boot集成RabbitMQ——高级特效——封装消息的元数据

交换机和队列支持持久化。现在我们也需要给消息设计元数据

DeliveryMode

设置为2，表示支持消息的持久化。

=======================================================================================

接上一边博文。  修改文件：

发送者：

package org.example.sender;
 
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageDeliveryMode;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
 
/**
 * 消息生产者 发送消息
 */
@Component
public class MessageSender {
 
    @Autowired
    RabbitTemplate rabbitTemplate;
 
    /**
     * 发送消息
     * @param info
     */
    public void send(String info)
    {
 
        System.out.println("发送消息>>>"+info);
 
//        CorrelationData correlationData = new CorrelationData();
//
//        String uuid = UUID.randomUUID().toString();
//        System.out.println(uuid);
//
//        correlationData.setId(uuid);
 
//        rabbitTemplate.convertAndSend("amqp-topic","huawei.a",info,correlationData);
 
        /**
         *     public static int toInt(MessageDeliveryMode mode) {
         *         switch (mode) {
         *         case NON_PERSISTENT:
         *             return 1;
         *         case PERSISTENT:
         *             return 2;
         *         default:
         *             return -1;
         */
 
 
        MessageProperties messageProperties = new MessageProperties();
 
        messageProperties.setDeliveryMode(MessageDeliveryMode.PERSISTENT);
 
        messageProperties.setExpiration("20000"); // 设置过期时间20秒
 
        messageProperties.setAppId("abc123456");
 
        messageProperties.setHeader("国家1","中国1");
        messageProperties.setHeader("国家2","中国2");
 
 
        Message message = new Message(info.getBytes(), messageProperties);
 
        System.out.println(message);
 
        rabbitTemplate.convertAndSend("amqp-topic", "huawei.a", message);
 
    }
 
}

 

 

消费者：

package org.example.receiver;
 
import com.rabbitmq.client.Channel;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
import java.io.IOException;
 
 
@Component
public class TopicReceiver {
 
    //分别监听名称为xiaomi、huawei的队列
    @RabbitListener(queues = "xiaomi")
    public void handlerXM(Message message,String msg, Channel channel) throws IOException {
 
        System.out.println("小米："+msg);
 
        //手动签收，不启动批量签收
        channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);
 
        System.out.println(message.getMessageProperties().getDeliveryTag());
    }
 
 
    @RabbitListener(queues = "huawei")
    public void handlerHW(Message message,String msg, Channel channel) throws IOException {
 
 
        System.out.println("华为："+msg);
 
        System.out.println(message.getMessageProperties().getHeaders());
 
        System.out.println((String) message.getMessageProperties().getHeader("国家2"));
 
        System.out.println(message);
 
        System.out.println(message.getMessageProperties().getExpiration());
 
        System.out.println(message.getMessageProperties().getAppId());
 
 
 
        //手动签收，不启动批量签收
        //告诉rmq签收的消息的id。以及是否批量签收
        channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);
 
    }
}

 

 

======================================================================================

参考：https://blog.csdn.net/qq_43623492/article/details/124259773