Spring集成RabbitMq的流程介绍

本是古典何须时尚 2022-05-09 09:18 174阅读 0赞

转自：[https://www.cnblogs.com/ityouknow/p/6120544.html][https_www.cnblogs.com_ityouknow_p_6120544.html]

package com.rabbit;
    
    /*
     * 完整的源码可以根据上面的链接看博文
     */
    
    @SpringBootApplication
    public class Application {
    
    
        //这样就声明了一个队列
        @Bean
        public Queue queueMessages() {
            return new Queue("topic.message");
        }
    
        //这样就声明了一个交换机
        @Bean
        TopicExchange exchange() {
            return new TopicExchange("exchange");
        }
       
        /**
         * 这样将队列topic.message与exchange绑定，binding_key为topic.message,就是完全匹配
         * 若有多个Queue,则取id为queueMessage的Queue
         * 若有多个TopicExchange,则取id为exchange的TopicExchange
         */
        @Bean
        Binding bindingExchangeMessage(Queue queueMessage,TopicExchange exchange){
        	return BindingBuilder.bind(queueMessage).to(exchange).with("topic.message");
        }
        
       
        public static void main(String[] args) throws Exception {
            SpringApplication.run(Application.class, args);
        }
    }

以上是声明了队列和交换机,并将队列和交换机绑定在一起

@Configuration  
    @PropertySource(value= {"classpath:application.properties"})
    public class RabbitConfig {
    	@Value("${spring.rabbitmq.host}")
    	private String addresses;
    
        //声明连接RabbitMq的ConnectionFactory
    	@Bean
    	public ConnectionFactory connectionFactory() {
    		CachingConnectionFactory connectionFactory = new CachingConnectionFactory();
    		connectionFactory.setAddresses(addresses + ":" + port);
    		connectionFactory.setUsername(username);
    		connectionFactory.setPassword(password);
    		connectionFactory.setVirtualHost(virtualHost);
    		/** 如果要进行消息回调，则这里必须要设置为true */
    		connectionFactory.setPublisherConfirms(publisherConfirms);
    		return connectionFactory;
    	}
    	/** 
             * RabbitTemplate用于发送消息
             *
    	 * 因为要设置回调类，所以应是prototype类型，如果是singleton类型，则回调类为最后一次设置,也就是多次调用send,回调类
    	 * 只会响应最后一次的send
    	 * ConfigurableBeanFactory.SCOPE_PROTOTYPE 每次注入的时候会自动创建一个新的bean实例
    	 * ConfigurableBeanFactory.SCOPE_SINGLETON 单例模式，在整个应用中只能创建一个实例
    	 * 
    	 *  一个RabbitTemplate只能支持一个ConfirmCallback,所以不能弄成Singleton
    	 *  当消息被brocker接收，broker会发送basic.ack，生产者可以通过回调函数确认这个消息
    	 *  如果因为RabbitMQ本身问题导致消息丢失，broker会发送basic.nack，生产者通过回调方法接收到nack后，可以考虑消息重发
    	 */
    	@Bean
    	@Scope(ConfigurableBeanFactory.SCOPE_PROTOTYPE)
    	public RabbitTemplate rabbitTemplatenew() {
    		RabbitTemplate template = new RabbitTemplate(connectionFactory());
    		return template;
    	}
    }

这里是声明了Rabbit的连接池跟RabbitTemplate

消息的消息有两种

一种是实现ChannelAwareMessageListener接口

一种是用@RabbitListener注解

区别在于用ChannelAwareMessageListener接口可以实现手工ACK，而@RabbitListener是自动ACK

@Configuration  
    public class RabbitConfig {
    	
    	@Bean(name="myListener")
    	public ChannelAwareMessageListener listener(){
    		return new AckTopicMessageReceiver();
    	}
    	
    	/**
    	 * 初始化监听器
    	 * 这里是监听Bean为queueMessages的Queue
    	 * SimpleMessageListenerContainer.setQueues方法可以监听多个Queue因为它的方法格式是setQueues(Queue... queues)
    	 */
    	@Bean
    	public SimpleMessageListenerContainer messageListener(ConnectionFactory connectionFactory,@Qualifier("queueMessages") Queue queues,@Qualifier("queueMessage") Queue queue,@Qualifier("myListener") ChannelAwareMessageListener listener){
    		SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory);
    		container.setAcknowledgeMode(AcknowledgeMode.MANUAL);
    		container.setQueues(queue,queues);
    		container.setMessageListener(listener);
    		return container;
    	}
    }

这里是声明了监听器,

public class AckTopicMessageReceiver implements ChannelAwareMessageListener{
    	public void onMessage(Message message, Channel channel) throws Exception {
    		String msg = new String(message.getBody());
    		long id = message.getMessageProperties().getDeliveryTag();
    		System.out.println("===接收到的消息:"+msg+",id："+id);
    		/*
    		 * 设置已消费消息,brocker可删除此消息
    		 * multiple为false只确认当前一个消息收到,true确认所有consumer获得的消息
    		 */
    		channel.basicAck(id,false);
    		/*
    		 * 第二参数是multiple
    		 * 第三参数是requeue,意思是要不要重新入队,true为重新入队,false为不重新入队,即丢弃这个意思
    		 */
    		channel.basicNack(id, false, true);
    	}
    }

[https_www.cnblogs.com_ityouknow_p_6120544.html]: https://www.cnblogs.com/ityouknow/p/6120544.html