Rabbitmq 使用及规范_rabbitmq 开发规范

EX.{交换机类型}.{模块名}.{功能名}.[业务名1].[业务名2]...

RK.{模块名}.{接收模块名}.{功能名}.[业务名1].[业务名2]...

package com.yuejf.utils.constant;

/**
 * redis key
 *
 * @author lu
 */
public class RabbitMqConstants {

    private RabbitMqConstants() {

    }

    /**
     * app业务
     */
    public static class APP {
        public static final String APP = "APP.";
        /****测试开始*****/
        public static final String TEST_EXCHANGE = TOPIC_EXCHANGE + APP + "TEST";
        public static final String TEST_QUEUE = QUEUE + APP + "TEST";
        /****测试结束*****/

    }

    /**
     * 支付业务
     */
    public static class PAY {

    }

    /**
     * 订单业务
     */
    public static class ORDER {

    }

    /**
     * 商品业务
     */
    public static class GOODS {

    }


    /**
     * 交换机
     */
    //直接类型
    public static final String TOPIC_EXCHANGE = "EX.TOPIC.";
    //主题类型
    public static final String DELAY_EXCHANGE = "EX.DELAY.";
    //标题类型
    public static final String FANOUT_EXCHANGE = "EX.FANOUT.";
    //扇出类型
    public static final String HEADERS_EXCHANGE = "EX.HEADERS.";
    /**
     * 队列 Queue
     */
    public static final String QUEUE = "MQ.";
    /**
     * Routing-key
     */
    public static final String ROUTING_KEY = "RK.";


}

@Configuration
public class RabbitConfig { 
   

    @Bean
    public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) { 
   
        final RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
        rabbitTemplate.setMessageConverter(jsonMessageConverter());
        return rabbitTemplate;
    }

    @Bean
    public Jackson2JsonMessageConverter jsonMessageConverter() { 
   
        return new Jackson2JsonMessageConverter();
    }
}

# 连接到RabbitMQ的虚拟主机 , 注意预上线环境为review
spring.rabbitmq.virtual-host=/test
#确认消息已发送到交换机(Exchange) 异步
spring.rabbitmq.publisher-confirm-type=orrelated
# 开启本地重试机制，默认3次，避免内存泄漏
spring.rabbitmq.listener.simple.retry.enabled=true
# 客户端连接的地址，有多个的时候使用逗号分隔，该地址可以是IP与Port的结合 
spring.rabbitmq.address       
# 当缓存已满时，获取Channel的等待时间，单位为毫秒  
spring.rabbitmq.cache.channel.checkout-timeout   
# 缓存中保持的Channel数量 
spring.rabbitmq.cache.channel.size    
# 连接缓存的模式 CHANNEL  
spring.rabbitmq.cache.connection.mode    
# 缓存的连接数 
spring.rabbitmq.cache.connection.size   
# 连接超时参数单位为毫秒：设置为“0”代表无穷大  
spring.rabbitmq.connnection-timeout    
# 默认创建一个AmqpAdmin的Bean true
spring.rabbitmq.dynamic    
# RabbitMQ的主机地址 localhost
spring.rabbitmq.host    
# 容器的acknowledge模式 
spring.rabbitmq.listener.acknowledge-mode    
# 启动时自动启动容器 true
spring.rabbitmq.listener.auto-startup 
# 消费者的最小数量    
spring.rabbitmq.listener.concurrency     
# 消费者每次从队列获取的消息数量 (默认一次250个)，配置消息的预读数量控制消费者一次性从队列中读取多少条消息，做到能者多劳的配置（因为在实际的生产环境中每个服务器的配置不可能完全相同，带来的处理消息的时间也不一样）。
spring.rabbitmq.listener.simple.prefetch
# 投递失败时是否重新排队 true
spring.rabbitmq.listener.default-requeue-rejected    
# 消费者的最大数量 
spring.rabbitmq.listener.max-concurrency    
# 在单个请求中处理的消息个数，他应该大于等于事务数量 
spring.rabbitmq.listener.prefetch     
# 不论是不是重试的发布 false  
spring.rabbitmq.listener.retry.enabled   
# 第一次与第二次投递尝试的时间间隔 1000
spring.rabbitmq.listener.retry.initial-interval   
# 尝试投递消息的最大数量 3
spring.rabbitmq.listener.retry.max-attempts   
# 两次尝试的最大时间间隔 10000
spring.rabbitmq.listener.retry.max-interval   
# 上一次尝试时间间隔的乘数 1.0
spring.rabbitmq.listener.retry.multiplier    
# 不论重试是有状态的还是无状态的 true
spring.rabbitmq.listener.retry.stateless    
# 在一个事务中处理的消息数量。为了获得最佳效果，该值应设置为小于等于每个请求中处理的消息个数，即spring.rabbitmq.listener.prefetch的值 
spring.rabbitmq.listener.transaction-size   
# 登录到RabbitMQ的密码  
spring.rabbitmq.password    
# RabbitMQ的端口号 5672
spring.rabbitmq.port    
# 开启Publisher Confirm机制 false
spring.rabbitmq.publisher-confirms    
# 开启publisher Return机制 false
spring.rabbitmq.publisher-returns   
# 请求心跳超时时间，单位为秒 
spring.rabbitmq.requested-heartbeat   
# 启用SSL支持 false
spring.rabbitmq.ssl.enabled   
# 保存SSL证书的地址 
spring.rabbitmq.ssl.key-store   
# 访问SSL证书的地址使用的密码  
spring.rabbitmq.ssl.key-store-password    
# SSL的可信地址 
spring.rabbitmq.ssl.trust-store    
#  访问SSL的可信地址的密码 
spring.rabbitmq.ssl.trust-store-password     
# SSL算法，默认使用Rabbit的客户端算法库 
spring.rabbitmq.ssl.algorithm    
# 启用强制信息 false。当mandatory标志位设置为true时，如果exchange根据自身类型和消息routingKey无法找到一个合适的queue存储消息，那么broker会调用basic.return方法将消息返还给生产者;当mandatory设置为false时，出现上述情况broker会直接将消息丢弃;通俗的讲，mandatory标志告诉broker代理服务器至少将消息route到一个队列中，否则就将消息return给发送者;   
spring.rabbitmq.template.mandatory   
# receive()方法的超时时间 0
spring.rabbitmq.template.receive-timeout    
# sendAndReceive()方法的超时时间 5000
spring.rabbitmq.template.reply-timeout    
# 设置为true的时候RabbitTemplate能够实现重试 false
spring.rabbitmq.template.retry.enabled    
# 第一次与第二次发布消息的时间间隔 1000
spring.rabbitmq.template.retry.initial-interval    
# 尝试发布消息的最大数量 3
spring.rabbitmq.template.retry.max-attempts    
# 尝试发布消息的最大时间间隔 10000
spring.rabbitmq.template.retry.max-interval    
# 上一次尝试时间间隔的乘数 1.0
spring.rabbitmq.template.retry.multiplier  
#  登录到RabbitMQ的用户名   
spring.rabbitmq.username   

@Configuration
public class DirectRabbitConfig {

    @Bean
    public Queue TestDirectQueue() {
        // durable:是否持久化,默认是false,持久化队列：会被存储在磁盘上，当消息代理重启时仍然存在，暂存队列：当前连接有效
        // exclusive:默认也是false，只能被当前创建的连接使用，而且当连接关闭后队列即被删除。此参考优先级高于durable
        // autoDelete:是否自动删除，当没有生产者或者消费者使用此队列，该队列会自动删除。
        //   return new Queue("TestDirectQueue",true,true,false);
        //一般设置一下队列的持久化就好,其余两个就是默认false
        return new Queue(RabbitMqConstants.APP.TEST_QUEUE, true);
    }

    //Direct交换机 起名：TestDirectExchange
    @Bean
    public DirectExchange TestDirectExchange() {
        return new DirectExchange(RabbitMqConstants.APP.TEST_EXCHANGE, true, false);
    }

    //绑定  将队列和交换机绑定, 并设置用于匹配键：TestDirectRouting
    @Bean
    public Binding bindingDirect() {
        return BindingBuilder.bind(TestDirectQueue()).to(TestDirectExchange()).with(RabbitMqConstants.APP.TEST_ROUTING_KEY);
    }
}


@RestController
@Slf4j
public class HelloController {

    @Autowired(required = false)
    private RabbitTemplate rabbitTemplate;

    @RequestMapping("/sendDirectConvertMessage")
    public R<?> sendDirectConvertMessage() {
        String messageId = String.valueOf(UUID.randomUUID());
        String messageData = "test convert message, hello!";
        String createTime = LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));

        Map<String, Object> map = new HashMap<>();
        map.put("messageId", messageId);
        map.put("messageData", messageData);
        map.put("createTime", createTime);
        //convertAndSend 方法相当于简化了的send方法,可以自动处理消息的序列化
        rabbitTemplate.convertAndSend(RabbitMqConstants.APP.TEST_EXCHANGE, RabbitMqConstants.APP.TEST_ROUTING_KEY, map);
        return R.ok();
    }

    @RequestMapping("/sendDirectMessage")
    public R<?> sendDirectMessage() {
        String messageId = String.valueOf(UUID.randomUUID());
        String messageData = "test message, hello!";
        String createTime = LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));

        Map<String, Object> map = new HashMap<>();
        map.put("messageId", messageId);
        map.put("messageData", messageData);
        map.put("createTime", createTime);

        //JackJson默认序列化
        ObjectMapper objectMapper = new ObjectMapper();
        Message message = MessageBuilder.withBody(objectMapper.toString().getBytes())
                .setContentType(MessageProperties.CONTENT_TYPE_TEXT_PLAIN)
                .setMessageId("1")
                .setHeader("header", "header")
                .build();
        rabbitTemplate.send(RabbitMqConstants.APP.TEST_EXCHANGE, RabbitMqConstants.APP.TEST_ROUTING_KEY, message);
        return R.ok();
    }
 }






@Slf4j
@Component
@RabbitListener(queues = RabbitMqConstants.APP.TEST_QUEUE)
public class TestMqListener {

    @RabbitHandler
    public void receiveMsg(String msg, Channel channel, Message message) throws IOException {
        log.info("Str收到的消息-> msg:{}", msg);
        try {
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        } catch (Exception ex) {
            log.error("TestMqListener#Exception:{}", ex);
            channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, true);
        }

    }

    @RabbitHandler
    public void receiveMsg(byte[] msg, Channel channel, Message message) throws IOException {
        log.info("By收到的消息-> msg:{}", msg.toString());
        try {
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        } catch (Exception ex) {
            log.error("TestMqListener#Exception:{}", ex);
            channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, true);
        }
    }

    @RabbitHandler
    public void receiveMsg(Map msg, Channel channel, Message message) throws IOException {
        log.info("Map收到的消息-> msg:{}", msg);
        try {
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        } catch (Exception ex) {
            log.error("TestMqListener#Exception:{}", ex);
            channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, true);
        }
    }

}