脚本之家 服务器常用软件
快捷导航
您的位置：首页 → 软件编程 → java → SpringBoot整合rabbitMq自定义消息转换

SpringBoot整合rabbitMq自定义消息转换方式

 更新时间：2023年09月28日 08:39:05   作者：YoungMirror  
这篇文章主要介绍了SpringBoot整合rabbitMq自定义消息转换方式,具有很好的参考价值,希望对大家有所帮助,如有错误或未考虑完全的地方,望不吝赐教

SpringBoot整合rabbitMq自定义消息转换

存入rabbitMq时转为string，取回来时转为对应的类

package com.medi.hk.conf;
import com.alibaba.fastjson.JSON;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.support.converter.AbstractMessageConverter;
import org.springframework.amqp.support.converter.MessageConversionException;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.context.annotation.Bean;
import org.springframework.http.MediaType;
import org.springframework.stereotype.Component;
@Component
@Slf4j
public class RabbitMqConfig {
    private final String mqConvertClassName = "className";
    /**
     * mq消息转换
     * @return
     */
    @Bean
    public MessageConverter messageConverter() {
        return new AbstractMessageConverter() {
            /**
             * 接收消息时转换
             * @param message
             * @return
             * @throws MessageConversionException
             */
            @Override
            public Object fromMessage(Message message) throws MessageConversionException {
                byte[] body = message.getBody();
                MessageProperties messageProperties = message.getMessageProperties();
                //需要转为对应类的类名
                String className = (String) messageProperties.getHeaders().get(mqConvertClassName);
                try {
                    Boolean aBoolean = (Boolean) messageProperties.getHeaders().get(isArray);
                    if (aBoolean != null && aBoolean) {
                        return JSONArray.parseArray(new String(body), Class.forName(className));
                    } else {
                        return JSON.parseObject(body, Class.forName(className));
                    }
                } catch (ClassNotFoundException e) {
                    log.error("mq转换错误: message ==> {} body ==> {}", JSON.toJSONString(message), new String(body), e);
                    throw new RuntimeException(e);
                }
            }
            /**
             * 发送消息时转换
             * @param object
             * @param messageProperties
             * @return
             */
            @Override
            protected Message createMessage(Object object, MessageProperties messageProperties) {
                messageProperties.setContentType(MediaType.APPLICATION_JSON_UTF8_VALUE);
                //标记时什么类转换过来的
                messageProperties.setHeader(mqConvertClassName, object.getClass().getName());
                //如果是数组，则设置数组成员的类型（取第一个，只允许类型相同的数组）
                if (object instanceof Collection) {
                    String className;
                    Collection coll = (Collection) object;
                    if (coll.size() > 0) {
                        Object next = coll.iterator().next();
                        className = next.getClass().getName();
                    } else {
                        className = Object.class.getName();
                    }
                    messageProperties.setHeader(mqConvertClassName, className);
                    messageProperties.setHeader(isArray, true);
                } else {
                    messageProperties.setHeader(mqConvertClassName, object.getClass().getName());
                }
                return new Message(JSON.toJSONBytes(object), messageProperties);
            }
        };
    }
}

springboot整合rabbitmq注意事项

随着springboot愈发广泛使用，各种中间件也需要和框架进行整合。

提示：以下是本篇文章正文内容，下面案例可供参考

记录本人在springboot下使用rabbitmq踩坑点。

在微服务架构下，生产者和消费者不在一个module下。消费者需要监听生产者发出的队列消息。

在github上clone的项目中，运行消费者模块时总是报错:队列404。

可是明明在生产者模块已经创建过。经过查阅资料，原因是生产者和消费者不在同一模块下导致。

解决方式有2种：

1.生产者消费者在同一模块下，即可避免

即:交换机、队列、绑定关系在同一模块下就可。

如图所示:

在这里插入图片描述

ExchangeConfig.java
package com.space.rbq.order.config;
import org.springframework.amqp.core.DirectExchange;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
/**
 * 消息交换机配置  可以配置多个
 * @author zhuzhe
 * @date 2018/5/25 15:40
 * @email 1529949535@qq.com
 */
@Configuration
public class ExchangeConfig {
    /** 消息交换机1的名字*/
    public final String EXCHANGE_01 = "first_exchange";
    /**
     *   1.定义direct exchange，绑定first_exchange
     *   2.durable="true" 持久化交换机， rabbitmq重启的时候不需要创建新的交换机
     *   3.direct交换器相对来说比较简单，匹配规则为：如果路由键匹配，消息就被投送到相关的队列
     *     fanout交换器中没有路由键的概念，他会把消息发送到所有绑定在此交换器上面的队列中。
     *     topic交换器你采用模糊匹配路由键的原则进行转发消息到队列中
     */
    @Bean
    public DirectExchange directExchange(){
        DirectExchange directExchange = new DirectExchange(EXCHANGE_01,true,false);
        return directExchange;
    }
}
QueueConfig
package com.space.rbq.order.config;
import org.springframework.amqp.core.Queue;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import java.util.HashMap;
import java.util.Map;
/**
 * 队列配置  可以配置多个队列
 * @author zhuzhe
 * @date 2018/5/25 13:25
 * @email 1529949535@qq.com
 */
@Configuration.java
public class QueueConfig {
    /*对列名称*/
    public static final String QUEUE_NAME1 = "first-queue";
    public static final String QUEUE_NAME2 = "second-queue";
    public static final String QUEUE_NAME3 = "third-queue";
    @Bean
    public Queue firstQueue() {
        /**
         durable="true" 持久化消息队列 ， rabbitmq重启的时候不需要创建新的队列
         auto-delete 表示消息队列没有在使用时将被自动删除 默认是false
         exclusive  表示该消息队列是否只在当前connection生效,默认是false
         */
        return new Queue(QUEUE_NAME1,true,false,false);
    }
    @Bean
    public Queue secondQueue() {
        return new Queue(QUEUE_NAME2,true,false,false);
    }
    @Bean
    public Queue thirdQueue() {
        // 配置 自动删除
        Map<String, Object> arguments = new HashMap<>();
        arguments.put("x-message-ttl", 60000);//60秒自动删除
        return new Queue(QUEUE_NAME3,true,false,true,arguments);
    }
}
RabbitMqConfig.java
package com.space.rbq.order.config;
import com.space.rbq.order.mqcallback.MsgSendConfirmCallBack;
import com.space.rbq.order.mqcallback.MsgSendReturnCallback;
import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import javax.annotation.PostConstruct;
/**
 * RabbitMq配置
 * @author zhuzhe
 * @date 2018/5/25 13:37
 * @email 1529949535@qq.com
 */
@Configuration
public class RabbitMqConfig {
    /**
     * key: queue在该direct-exchange中的key值，当消息发送给direct-exchange中指定key为设置值时，
     * 消息将会转发给queue参数指定的消息队列
     */
    /** 队列key1*/
    public static final String ROUTING_KEY_1 = "queue_one_key1";
    public static final String ROUTING_KEY_2 = "queue_one_key2";
    @Autowired
    private QueueConfig queueConfig;
    @Autowired
    private ExchangeConfig exchangeConfig;
    @Autowired
    private ConnectionFactory connectionFactory;
//    @Autowired
//    RabbitTemplate rabbitTemplate;
//    @PostConstruct
//    public void init() {
//        rabbitTemplate.setConfirmCallback(msgSendConfirmCallBack());
//        rabbitTemplate.setReturnCallback(msgSendReturnCallback());
//    }
    /**
     * 将消息队列1和交换机1进行绑定,指定队列key1
     */
    @Bean
    public Binding binding_one() {
        return BindingBuilder.bind(queueConfig.firstQueue()).to(exchangeConfig.directExchange()).with(RabbitMqConfig.ROUTING_KEY_1);
    }
    /**
     * 将消息队列2和交换机1进行绑定,指定队列key2
     */
    @Bean
    public Binding binding_two() {
        return BindingBuilder.bind(queueConfig.secondQueue()).to(exchangeConfig.directExchange()).with(RabbitMqConfig.ROUTING_KEY_2);
    }
    /**
     * 定义rabbit template用于数据的接收和发送
     * 可以设置消息确认机制和回调
     * @return
     */
    @Bean
    public RabbitTemplate rabbitTemplate() {
        RabbitTemplate template = new RabbitTemplate(connectionFactory);
        // template.setMessageConverter(); 可以自定义消息转换器  默认使用的JDK的，所以消息对象需要实现Serializable
        // template.setMessageConverter(new Jackson2JsonMessageConverter());
        /**若使用confirm-callback或return-callback，
         * 必须要配置publisherConfirms或publisherReturns为true
         * 每个rabbitTemplate只能有一个confirm-callback和return-callback
         */
        template.setConfirmCallback(msgSendConfirmCallBack());
        /**
         * 使用return-callback时必须设置mandatory为true，或者在配置中设置mandatory-expression的值为true，
         * 可针对每次请求的消息去确定'mandatory'的boolean值，
         * 只能在提供'return -callback'时使用，与mandatory互斥
         */
        template.setReturnCallback(msgSendReturnCallback());
        template.setMandatory(true);
        return template;
    }
    /**
     * 关于 msgSendConfirmCallBack 和 msgSendReturnCallback 的回调说明：
     * 1.如果消息没有到exchange,则confirm回调,ack=false
     * 2.如果消息到达exchange,则confirm回调,ack=true
     * 3.exchange到queue成功,则不回调return
     * 4.exchange到queue失败,则回调return(需设置mandatory=true,否则不回调,消息就丢了)
     */
    /**
     * 消息确认机制
     * Confirms给客户端一种轻量级的方式，能够跟踪哪些消息被broker处理，
     * 哪些可能因为broker宕掉或者网络失败的情况而重新发布。
     * 确认并且保证消息被送达，提供了两种方式：发布确认和事务。(两者不可同时使用)
     * 在channel为事务时，不可引入确认模式；同样channel为确认模式下，不可使用事务。
     * @return
     */
    @Bean
    public MsgSendConfirmCallBack msgSendConfirmCallBack(){
        return new MsgSendConfirmCallBack();
    }
    @Bean
    public MsgSendReturnCallback msgSendReturnCallback(){
        return new MsgSendReturnCallback();
    }
}

消费者监听器配置:

在这里插入图片描述

 /**
     * queues  指定从哪个队列（queue）订阅消息
     * @param message
     * @param channel
     */
    @RabbitListener(queues = {QUEUE_NAME1})
    public void handleMessage(Message message,Channel channel) throws IOException {
        try {
            // 处理消息
            System.out.println("OrderConsumer {} handleMessage :"+message);
            // 执行减库存操作
//            storeService.update(new Gson().fromJson(new String(message.getBody()),Order.class));
           /**
            * 第一个参数 deliveryTag：就是接受的消息的deliveryTag,可以通过msg.getMessageProperties().getDeliveryTag()获得
            * 第二个参数 multiple：如果为true，确认之前接受到的消息；如果为false，只确认当前消息。
            * 如果为true就表示连续取得多条消息才发会确认，和计算机网络的中tcp协议接受分组的累积确认十分相似，
            * 能够提高效率。
            *
            * 同样的，如果要nack或者拒绝消息（reject）的时候，
            * 也是调用channel里面的basicXXX方法就可以了（要指定tagId）。
            *
            * 注意：如果抛异常或nack（并且requeue为true），消息会重新入队列，
            * 并且会造成消费者不断从队列中读取同一条消息的假象。
            */
            // 确认消息
            // 如果 channel.basicAck   channel.basicNack  channel.basicReject 这三个方法都不执行，消息也会被确认 【这个其实并没有在官方看到，不过自己测试的确是这样哈】
            // 所以，正常情况下一般不需要执行 channel.basicAck
            // channel.basicAck(message.getMessageProperties().getDeliveryTag(), true);
        }catch (Exception e){
            log.error("OrderConsumer  handleMessage {} , error:",message,e);
            // 处理消息失败，将消息重新放回队列
            channel.basicNack(message.getMessageProperties().getDeliveryTag(), false,true);
        }
    }

2.使用@RabbitListener 动态绑定

RabbitListener(bindings = @QueueBinding(
            value = @Queue(value = QUEUE_NAME1, durable = "true", autoDelete = "false"),
            exchange = @Exchange(value = "first_exchange", type = ExchangeTypes.DIRECT),
            key = "queue_one_key1"))

注意：

在微服务架构下吗，如果消息发送方和消费方不在一个模块下，需要使用以上两种方式。

总结

以上为个人经验，希望能给大家一个参考，也希望大家多多支持脚本之家。

您可能感兴趣的文章:

相关文章

  • Mybatis之@ResultMap,@Results,@Result注解的使用

    Mybatis之@ResultMap,@Results,@Result注解的使用

    这篇文章主要介绍了Mybatis之@ResultMap,@Results,@Result注解的使用，具有很好的参考价值，希望对大家有所帮助。如有错误或未考虑完全的地方，望不吝赐教
    2021-12-12
  • SpringBoot中的FailureAnalyzer使用详解

    SpringBoot中的FailureAnalyzer使用详解

    这篇文章主要介绍了SpringBoot中的FailureAnalyzer使用详解,Spring Boot的FailureAnalyzer是一个接口,它用于在Spring Boot应用启动失败时提供有关错误的详细信息,这对于开发者来说非常有用,因为它可以帮助我们快速识别问题并找到解决方案,需要的朋友可以参考下
    2023-12-12
  • 解决IDEA克隆代码后在右下角没有git分支的问题

    解决IDEA克隆代码后在右下角没有git分支的问题

    这篇文章主要介绍了解决IDEA克隆代码后在右下角没有git分支的问题，具有很好的参考价值，希望对大家有所帮助。一起跟随小编过来看看吧
    2021-02-02
  • Spring Boot优化后启动速度快到飞起技巧示例

    Spring Boot优化后启动速度快到飞起技巧示例

    这篇文章主要为大家介绍了Spring Boot优化后启动速度快到飞起的技巧示例详解，有需要的朋友可以借鉴参考下，希望能够有所帮助，祝大家多多进步，早日升职加薪
    2022-07-07
  • Java中final关键字的使用与注意总结

    Java中final关键字的使用与注意总结

    这篇文章主要给大家介绍了关于Java中final关键字的相关资料，文中通过示例代码介绍的非常详细，对大家的学习或者使用Java具有一定的参考学习价值，需要的朋友们下面来一起学习学习吧
    2020-08-08
  • Java开发环境配置方法

    Java开发环境配置方法

    这篇文章主要介绍了并不是每一台计算机都可以运行Java程序，要运行Java程序，计算机必须学习搭建Java开发环境,需要的朋友可以参考下
    2015-07-07
  • Java Spring事务使用及验证过程详解

    Java Spring事务使用及验证过程详解

    这篇文章主要介绍了Java Spring事务使用及验证过程详解,文中通过示例代码介绍的非常详细，对大家的学习或者工作具有一定的参考学习价值,需要的朋友可以参考下
    2020-12-12
  • Mybatis之映射实体类中不区分大小写的解决

    Mybatis之映射实体类中不区分大小写的解决

    这篇文章主要介绍了Mybatis之映射实体类中不区分大小写的解决，具有很好的参考价值，希望对大家有所帮助。如有错误或未考虑完全的地方，望不吝赐教
    2021-11-11
  • java实现简单控制台五子棋游戏

    java实现简单控制台五子棋游戏

    这篇文章主要为大家详细介绍了java实现简单控制台五子棋游戏，文中示例代码介绍的非常详细，具有一定的参考价值，感兴趣的小伙伴们可以参考一下
    2019-11-11
  • JBoss5.x下配置Log4j方法介绍

    JBoss5.x下配置Log4j方法介绍

    这篇文章主要介绍了JBoss5.x下配置Log4j方法介绍，小编觉得挺不错的，这里分享给大家，供需要的朋友参考。
    2017-10-10

最新评论