集群与负载均衡系列(5)——消息队列之spring-boot整合Rabbitmq

小鱼儿 2022-06-16 08:15 256阅读 0赞

上一篇文章介绍了如何搭建Rabbitmq，这篇文章说一下如何和spring-boot进行整合。这里主要用java config，当然你也可以用xml或者声明式注释配置

*         依赖

<dependency>  
            <groupId>org.springframework.boot</groupId>  
            <artifactId>spring-boot-starter-amqp</artifactId>  
    </dependency>

*          yml配置

注意，这里有黄色下划线的项说明spring-boot没有提供自动配置，我这里是通过@ConfigurationProperties进行处理的。你也可以不这样处理

![Center][]

*           转换为properties对象

你可以不像我这样处理，可以用@Value，静态成员变量都可以

package com.wlf.demo.props;
    
    import java.util.Map;
    
    import org.springframework.boot.context.properties.ConfigurationProperties;
    import org.springframework.stereotype.Component;
    
    @Component  
    @ConfigurationProperties(prefix="spring.rabbitmq")
    public class RabbitmqProps {
    
    	private String addresses;
    	private String username;
    	private String password;
    	private boolean publisherConfirms;
    	private String exchange;
    	private String queueName;
    	private Map<String,String> keys;
    	
    	public String getAddresses() {
    		return addresses;
    	}
    	public void setAddresses(String addresses) {
    		this.addresses = addresses;
    	}
    	public String getUsername() {
    		return username;
    	}
    	public void setUsername(String username) {
    		this.username = username;
    	}
    	public String getPassword() {
    		return password;
    	}
    	public void setPassword(String password) {
    		this.password = password;
    	}
    	public boolean isPublisherConfirms() {
    		return publisherConfirms;
    	}
    	public void setPublisherConfirms(boolean publisherConfirms) {
    		this.publisherConfirms = publisherConfirms;
    	}
    	public String getExchange() {
    		return exchange;
    	}
    	public void setExchange(String exchange) {
    		this.exchange = exchange;
    	}
    	public String getQueueName() {
    		return queueName;
    	}
    	public void setQueueName(String queueName) {
    		this.queueName = queueName;
    	}
    	public Map<String, String> getKeys() {
    		return keys;
    	}
    	public void setKeys(Map<String, String> keys) {
    		this.keys = keys;
    	}
    	
    }

*            java config

主要配置连接，队列，交换机，绑定规则，RabbitTemplate和消费端监听。如果不适用消费端的监听，可以用注释中的SimpleMessageListenerContainer替换SimpleRabbitListenerContainerFactory。

package com.wlf.demo;
    
    import org.springframework.amqp.core.AcknowledgeMode;
    import org.springframework.amqp.core.Binding;
    import org.springframework.amqp.core.BindingBuilder;
    import org.springframework.amqp.core.DirectExchange;
    
    import org.springframework.amqp.core.Queue;
    import org.springframework.amqp.rabbit.annotation.EnableRabbit;
    import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
    import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
    import org.springframework.amqp.rabbit.connection.ConnectionFactory;
    import org.springframework.amqp.rabbit.core.ChannelAwareMessageListener;
    import org.springframework.amqp.rabbit.core.RabbitTemplate;
    import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
    import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
    import org.springframework.context.annotation.Scope;
    
    import com.rabbitmq.client.Channel;
    import com.wlf.demo.props.RabbitmqProps;
    
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.beans.factory.config.ConfigurableBeanFactory;
    import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
    import org.springframework.boot.autoconfigure.SpringBootApplication;
    import org.springframework.boot.context.properties.ConfigurationProperties;
    
    @SpringBootApplication
    @EnableAutoConfiguration
    @EnableRabbit //使用@RabbitListener必须加该注释
    public class AmqpConfig {
    	
    	@Autowired
    	private RabbitmqProps rabbitmqProps;
    
    	@Bean
    	@ConfigurationProperties(prefix="spring.rabbitmq")   
        public ConnectionFactory connectionFactory() {  
            CachingConnectionFactory connectionFactory = new CachingConnectionFactory();  
            connectionFactory.setAddresses(rabbitmqProps.getAddresses());  
            connectionFactory.setUsername(rabbitmqProps.getUsername());  
            connectionFactory.setPassword(rabbitmqProps.getPassword());  
            connectionFactory.setVirtualHost("/");  
            connectionFactory.setPublisherConfirms(rabbitmqProps.isPublisherConfirms()); //消息回调，必须要设置  
            return connectionFactory;  
        }  
    	
        //必须是prototype类型  
        @Bean  
        @Scope(ConfigurableBeanFactory.SCOPE_PROTOTYPE)  
        public RabbitTemplate rabbitTemplate() {  
            RabbitTemplate template = new RabbitTemplate(connectionFactory());  
            return template;  
        }  
    	
        /**  
         * 针对消费者配置  
         * 1. 设置交换机类型  
         * 2. 将队列绑定到交换机  
         *   
         *   
            FanoutExchange: 将消息分发到所有的绑定队列，无routingkey的概念  
            HeadersExchange ：通过添加属性key-value匹配  
            DirectExchange:按照routingkey分发到指定队列  
            TopicExchange:多关键字匹配  
         */  
    
        @Bean  
        public DirectExchange defaultExchange() {  
            return new DirectExchange(rabbitmqProps.getExchange());  
        }
    
        @Bean  
        public Queue queue() {  
            return new Queue(rabbitmqProps.getQueueName(), true); //队列持久  
        }  
    
        @Bean  
        public Binding binding() {  
            return BindingBuilder.bind(queue()).to(defaultExchange()).with(rabbitmqProps.getKeys().get("orderBounding"));  
        }  
    
        /*
        @Bean  
        public SimpleMessageListenerContainer messageContainer() {  
            SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory());  
            container.setQueues(queue());  
            container.setExposeListenerChannel(true);  
            container.setMaxConcurrentConsumers(10);  
            container.setConcurrentConsumers(3);  
            container.setAcknowledgeMode(AcknowledgeMode.MANUAL); //设置确认模式手工确认
            container.setMessageListener(new ChannelAwareMessageListener(){
    
    			public void onMessage(Message message, Channel channel) throws Exception {
    				byte[] body = message.getBody();  
                    System.out.println("receive msg : " + new String(body));  
                    channel.basicAck(message.getMessageProperties().getDeliveryTag(), false); //确认消息成功消费  
    			}
            	
            });
    
            return container;  
        }  
        */
        
        //使用@RabbitListener必须配置该选项
        @Bean
        public SimpleRabbitListenerContainerFactory rabbitListenerContainerFactory() {
            SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
            factory.setConnectionFactory(connectionFactory());
            factory.setConcurrentConsumers(3);
            factory.setMaxConcurrentConsumers(10);
            return factory;
        }
        
    }

到此为止，已经将RabbitMq整合进spring-boot了

*         示例

接下来简单写个使用的示例。模仿订单处理，用户下订单后，将消息发送给队列，马上返回用户订单已下达。消费端进行订单处理，如果存在问题，通过一定的方式通知用户，比如短信的方式。

在实际中，可能一个生成端对应多个消费端，消费端也可以通过之前nginx的方式集群以及负载均衡。另外，RabbitMq的一对多中已经实现了负载均衡。

控制层提供下达订单的接口

package com.wlf.demo.controller;
    
    import javax.servlet.http.HttpServletRequest;
    import javax.servlet.http.HttpServletResponse;
    
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.http.MediaType;
    import org.springframework.web.bind.annotation.PathVariable;
    import org.springframework.web.bind.annotation.RequestMapping;
    import org.springframework.web.bind.annotation.RequestMethod;
    import org.springframework.web.bind.annotation.RestController;
    
    import com.wlf.demo.bussiness.OrderService;
    
    @RestController
    public class OrderController {
    
    	@Autowired
    	private OrderService orderService;
    	
    	@RequestMapping(value = "/rest/order/{id}",
    			method = RequestMethod.POST, 
    			produces = MediaType.TEXT_PLAIN_VALUE+";charset=UTF-8")
    	public void saveOrder(@PathVariable String id,HttpServletRequest request,HttpServletResponse response){
    		orderService.saveOrder(id);
    	}
    	
    }

服务端将订单编号发到队列，并且马上返回用户订单已下达。这里使用了确认机制，该机制也是异步的，它将消息是否发送到队列的情况进行回调。这里失败或超时后没有做重新发送处理

package com.wlf.demo.bussiness;
    
    import org.springframework.stereotype.Service;
    
    import com.wlf.demo.props.RabbitmqProps;
    
    import java.util.UUID;
    import org.springframework.amqp.rabbit.core.RabbitTemplate;
    import org.springframework.amqp.rabbit.support.CorrelationData;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.beans.factory.annotation.Value; 
    
    @Service("orderService")
    public class OrderService implements RabbitTemplate.ConfirmCallback {
    	
    	@Value("${spring.rabbitmq.keys.orderBounding}")
    	private String boundingKey;
    	
    	@Value("${spring.rabbitmq.exchange}")
    	private String testExchange;
    	
    	@Autowired
    	private RabbitmqProps rabbitmqProps;
    	
    	private RabbitTemplate rabbitTemplate;
    
    	/**
    	 * 发布确认机制，生产者发送消息后的回调.也就是是否可路由
    	 */
    	public void confirm(CorrelationData correlationData, boolean ack, String cause) {
    
    		System.out.println(" 回调id:" + correlationData);  
            if (ack) {  
                System.out.println("订单发送成功！");  
            } else {  
                System.out.println("订单发送失败，进行重新发送" + cause);  
            }  
    		
    	}
    
    	@Autowired  
        public OrderService(RabbitTemplate rabbitTemplate) {  
            this.rabbitTemplate = rabbitTemplate;  
            rabbitTemplate.setConfirmCallback(this); 
        }  
    	
    	public void saveOrder(String id){
    		rabbitTemplate.setRoutingKey(rabbitmqProps.getKeys().get("orderRouting"));
    		CorrelationData correlationId = new CorrelationData(UUID.randomUUID().toString());
    		rabbitTemplate.convertAndSend(testExchange, boundingKey, id, correlationId);
    	}
    	
    }

消费端处理订单，将订单持久化到数据库，如果处理失败，通过一定方式通知用户。可以在持久化到数据库的时候做去重处理。

package com.wlf.demo.bussiness;
    
    import org.springframework.amqp.rabbit.annotation.RabbitHandler;
    import org.springframework.amqp.rabbit.annotation.RabbitListener;
    import org.springframework.stereotype.Component;
    
    @Component
    @RabbitListener(queues = "${spring.rabbitmq.queueName}")
    public class OrderRecv {
    
    	@RabbitHandler
        public void receive(String message) {
    		try{
    			System.out.println("收到信息: " + message);
    			//订单数据库处理
    			System.out.println("订单已存储到数据库: " + message);
    		}catch(Exception e){
    			System.out.println("发送失败信息给用户邮箱，或发送短信，推送消息");
    		}
        }
    	
    }

最后是单元测试代码

package com.wlf.demo;
    
    import org.apache.http.client.methods.HttpPost;
    import org.apache.http.impl.client.CloseableHttpClient;
    import org.apache.http.impl.client.HttpClients;
    import org.junit.Test;
    
    public class OrderTest {
    
    	@Test
    	public void loginTest() throws Exception{
    		String url="http://localhost:8080/rest/order/order_123_666_888";
    		CloseableHttpClient httpclient=HttpClients.createDefault();
            HttpPost httppost=new HttpPost(url);
            httpclient.execute(httppost);
    	}
    	
    }

示例代码地址 [https://github.com/wulinfeng2/spring-boot-rabbitmq][https_github.com_wulinfeng2_spring-boot-rabbitmq]

[Center]: /images/20220616/7c4719193ae94186bd590ade1efc198e.png
[https_github.com_wulinfeng2_spring-boot-rabbitmq]: https://github.com/wulinfeng2/spring-boot-rabbitmq