RabbitMQ消息监听（多种模式

1.rabbitmq消息监听，兼容多种模式的消息，fanout/topic等模式

MQ消息配置监听：

1package com.test.ddyin.conf;
2
3import java.util.HashMap;
4import java.util.List;
5import java.util.function.Predicate;
6import java.util.stream.Collectors;
7
8import org.apache.poi.ss.formula.functions.T;
9import org.springframework.amqp.core.AbstractExchange;
10import org.springframework.amqp.core.Binding;
11import org.springframework.amqp.core.Binding.DestinationType;
12import org.springframework.amqp.core.BindingBuilder;
13import org.springframework.amqp.core.Exchange;
14import org.springframework.amqp.core.FanoutExchange;
15import org.springframework.amqp.core.Queue;
16import org.springframework.amqp.core.TopicExchange;
17import org.springframework.amqp.rabbit.connection.ConnectionFactory;
18import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
19import org.springframework.amqp.rabbit.listener.adapter.MessageListenerAdapter;
20import org.springframework.beans.factory.annotation.Autowired;
21import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
22import org.springframework.context.annotation.Bean;
23import org.springframework.context.annotation.Configuration;
24
25import com.qf.openchannel.mq.MQMessageAware;
26import com.qf.openchannel.mq.MQReceiver;
27
28@Configuration
29@ConditionalOnProperty(prefix = "spring.rabbitmq", name = "enable", matchIfMissing = false)
30public class MQInitConfig {
31
32	private final String queueNameSufix = ".test.channel";
33	@Autowired
34	private List<MQMessageAware> messageListeners;
35
36	@Bean
37	List<Queue> queue() {
38		return messageListeners.stream().map(listener -> {
39			return new Queue(listener.getExchange() + queueNameSufix, false);
40		}).collect(Collectors.toList());
41	}
42
43	@Bean
44	List<Exchange> exchange() {
45		return messageListeners.stream().map(listener -> {
46			return new AbstractExchange(listener.getExchange()) {
47				@Override
48				public String getType() {
49					return listener.getMQType();
50				}
51			};
52		}).collect(Collectors.toList());
53	}
54
55	@Bean
56	List<Binding> binding() {
57		return messageListeners.stream().map(listener -> {
58			return new Binding(listener.getExchange() + queueNameSufix, DestinationType.QUEUE, listener.getExchange(),
59					listener.getRoutingKey(), new HashMap<String, Object>());
60		}).collect(Collectors.toList());
61	}
62
63	@Bean
64	SimpleMessageListenerContainer container(ConnectionFactory connectionFactory,
65			MessageListenerAdapter listenerAdapter) {
66		SimpleMessageListenerContainer container = new SimpleMessageListenerContainer();
67		container.setConnectionFactory(connectionFactory);
68		container.setMessageListener(listenerAdapter);
69		messageListeners.forEach(listener -> {
70			container.addQueueNames(listener.getExchange() + queueNameSufix);
71		});
72		return container;
73	}
74
75	@Bean
76	MessageListenerAdapter listenerAdapter(MQReceiver receiver) {
77		return new MessageListenerAdapter(receiver);
78	}
79}

注意：在绑定的时候，要加入exchange和routing key（fanout模式的routing key 为空字符串），其中，在queue，exchange和binding加注解，相当于是在容器中添加了exchange，队列和两者之间的绑定关系，可以直接从applicationContext中获取，其中，也是自动创建了exchange，queue以及两者之间的绑定关系，不需要在rabbitmq界面重新添加exchange，queue以及两者的绑定关系。

MQ消息接收：(MQReceiver)

1package com.qf.openchannel.mq;
2
3import java.util.HashMap;
4import java.util.Map;
5
6import org.springframework.amqp.core.Message;
7import org.springframework.amqp.core.MessageListener;
8import org.springframework.beans.BeansException;
9import org.springframework.context.ApplicationContext;
10import org.springframework.context.ApplicationContextAware;
11import org.springframework.context.ApplicationListener;
12import org.springframework.context.event.ContextRefreshedEvent;
13import org.springframework.stereotype.Service;
14
15import com.qf.openchannel.util.Constant;
16import com.qf.openchannel.util.LoggerUtil;
17
18@Service
19public class MQReceiver implements MessageListener, ApplicationContextAware, ApplicationListener<ContextRefreshedEvent> {
20	
21	private final String queueNameSufix = ".test.channel";
22	private ApplicationContext applicationContext;
23	private Map<String, MQMessageAware> messageListener = new HashMap<>();
24
25	@Override
26	public void onMessage(Message message) {
27		String payload = new String(message.getBody());
28		LoggerUtil.info("Received <" + payload + ">");
29		try {			
30			String exchange = message.getMessageProperties().getConsumerQueue();
31			String routingKey = message.getMessageProperties().getReceivedRoutingKey();
32			LoggerUtil.info("MQReceiverService onMessage routingKey {} exchange {}", routingKey,exchange);
33			if (messageListener.containsKey(exchange)) {
34				if(messageListener.containsKey(routingKey)) {
35					messageListener.get(routingKey).onMessage(payload);
36				}else {
37					messageListener.get(exchange).onMessage(payload);
38				}
39			} else {
40				LoggerUtil.info("MQReceiverService receiveMessage unrecognized message from exchange : ", exchange);
41			}
42		} catch (Exception e) {
43			LoggerUtil.error("MQReceiverService receiveMessage exception: ", e);
44		}
45	}
46
47	@Override
48	public void onApplicationEvent(ContextRefreshedEvent event) {
49		applicationContext.getBeansOfType(MQMessageAware.class).forEach((key, listener) -> {
50			if(listener.getMQType().equals(Constant.MQTYPE_TOPIC)) {
51				LoggerUtil.info("MQReceiverService receiveMessage messageType {} routingKey {} exchange {}", listener.getMQType(), listener.getRoutingKey(),listener.getExchange());
52				messageListener.put(listener.getExchange() + queueNameSufix, listener);
53				messageListener.put(listener.getRoutingKey(), listener);
54			}else if(listener.getMQType().equals(Constant.MQTYPE_FANOUT)){
55				LoggerUtil.info("MQReceiverService receiveMessage messageType {} routingKey {} exchange {}", listener.getMQType(), listener.getRoutingKey(),listener.getExchange());
56				messageListener.put(listener.getExchange() + queueNameSufix, listener);
57			}else {
58				
59			}
60		});
61	}
62
63	@Override
64	public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
65		this.applicationContext = applicationContext;
66	}
67}

注意：区分不同消息类型来绑定不同的监听，对于topic模式，routing key也要绑定对应的listener（监听器），然后通过message可获取监听的exchange和routing key

对于监听器，由于有多个监听，抽象出一个共同接口：

MQMessageAware

1package com.test.ddyin.mq;
2
3public interface MQMessageAware {
4	String getExchange();
5	void onMessage(String message);
6	String getMQType();
7	String getRoutingKey();
8}

然后对于不同的监听可手动实现：

例如：退团消息的监听：

1package com.test.ddyin.mq;
2
3import org.springframework.beans.factory.annotation.Autowired;
4import org.springframework.stereotype.Service;
5
6import com.fasterxml.jackson.databind.ObjectMapper;
7import com.qf.openchannel.model.QuitPlan;
8import com.qf.openchannel.service.QuitPlanService;
9import com.qf.openchannel.util.Constant;
10import com.qf.openchannel.util.DateUtil;
11import com.qf.openchannel.util.LoggerUtil;
12
13/**
14 * @author ddyin
15 * @date 2017年9月8日 下午14:08:34
16 */
17@Service
18public class QuitPlanListener implements MQMessageAware{
19	
20	@Autowired
21	QuitPlanService quitPlanService;
22
23	@Override
24	public String getExchange() {
25		return "trade.topic.notification";
26	}
27
28	@Override
29	public void onMessage(String message) {
30		try {
31			LoggerUtil.info("QuitPlanListener dealMessage start: {}", DateUtil.get14Date());
32			ObjectMapper mapper = new ObjectMapper();
33			QuitPlan quitPlan = mapper.readValue(message, QuitPlan.class);
34			quitPlanService.insertQuitPlan(quitPlan);
35			LoggerUtil.info("QuitPlanListener dealMessage end: {}", DateUtil.get14Date());
36		} catch (Exception e) {
37			LoggerUtil.error("QuitPlanListener.onMessage Exception:{}", e);
38		}
39	}
40
41	@Override
42	public String getMQType() {
43		return Constant.MQTYPE_TOPIC;
44	}
45	
46	@Override
47	public String getRoutingKey() {
48		return "trade.plan.status.settled";
49	}
50	
51}

到此，rabbitmq监听可实现不同消息类型的监听。

注意项目中rabbitmq的配置：

1rabbitmq:
2    host: 6.6.6.6
3    port: 5674
4    username: test
5    password: test
6    virtual-host: /test
7    enable: false

综述，end

补充:

如果想扩展到多个virtualHost，可以添加ConnectionFactory

其中配置的virtualHost配置在配置文件中，目的是区分对接不同的业务，通过virtualHost来进行隔离。

事例如下:(放置在MqInitConfig.java文件中)

1    /**
2     * virtual-host: /host1 ConnectionFactory
3     *
4     * @return
5     */
6    @Bean
7    ConnectionFactory connectionFactory1() {
8        com.rabbitmq.client.ConnectionFactory connectionFactory = new com.rabbitmq.client.ConnectionFactory();
9        connectionFactory.setHost(mqConfig.getHost());
10        connectionFactory.setPort(mqConfig.getPort());
11        connectionFactory.setUsername(mqConfig.getUsername());
12        connectionFactory.setPassword(mqConfig.getPassword());
13        connectionFactory.setVirtualHost(mqConfig.getVirtualHost1());
14
15        CachingConnectionFactory factory = new CachingConnectionFactory(connectionFactory);
16        return factory;
17    }
18
19    /**
20     * virtual-host: /host2 ConnectionFactory
21     *
22     * @return
23     */
24    @Bean
25    ConnectionFactory connectionFactory2() throws IOException, TimeoutException {
26        com.rabbitmq.client.ConnectionFactory connectionFactory = new com.rabbitmq.client.ConnectionFactory();
27        connectionFactory.setHost(mqConfig.getHost());
28        connectionFactory.setPort(mqConfig.getPort());
29        connectionFactory.setUsername(mqConfig.getUsername());
30        connectionFactory.setPassword(mqConfig.getPassword());
31        connectionFactory.setVirtualHost(mqConfig.getVirtualHost2());
32
33
34        CachingConnectionFactory factory = new CachingConnectionFactory(connectionFactory);
35        return factory;
36    }

添加完之后将多个virtualHost加入到SimpleRoutingConnectionFactory

1@Bean
2    ConnectionFactory connectionFactory() {
3        SimpleRoutingConnectionFactory factory = new SimpleRoutingConnectionFactory();
4        Map<Object, ConnectionFactory> targetConnectionFactories = new HashMap<>();
5        targetConnectionFactories.put("connectionFactory1", connectionFactory1());
6        try {
7            targetConnectionFactories.put("connectionFactory2", connectionFactory2());
8        } catch (IOException e) {
9            LoggerUtil.error("connectionFactory targetConnectionFactories IOException: {}", e);
10        } catch (TimeoutException e) {
11            LoggerUtil.error("connectionFactory targetConnectionFactories TimeoutException: {}", e);
12        }
13        factory.setTargetConnectionFactories(targetConnectionFactories);
14        return factory;
15    }

可以将对应的connectionFactory添加到container中，通过virtualHost来进行区分。

1    @Bean
2    SimpleMessageListenerContainer container(ConnectionFactory connectionFactory,
3                                              MessageListenerAdapter listenerAdapter) {
4        SimpleMessageListenerContainer container = new SimpleMessageListenerContainer();
5        container.setConnectionFactory(connectionFactory);
6        container.setMessageListener(listenerAdapter);
7        messageListeners.forEach(listener -> {
8            if (MQ_RABBIT_VIRTUAL_HOST2.equals(listener.getVirtualHost())) {
9                container.addQueueNames(listener.getExchange() + queueNameSufix);
10            }else{
11                //container.addQueueNames(listener.getExchange() + queueNameSufix);
12                //或者其他业务逻辑
13            }
14        });
15        return container;
16    }

当然也要在MQMessageAware接口中添加方法：

1public interface MQMessageAware {
2	String getExchange();
3	void onMessage(String message);
4	String getMQType();
5	String getRoutingKey();
6	String getVirtualHost();
7}

可实现多种virtualHost多种配置。。。

补充:

1.当消费者消费信息出现异常时，比如消费者宕机，消息该如何处理，当生产者宕机时，消息该如何处理?

A:对于消费者宕机，rabbitmq提供ack机制，当ack机制设置成true的时候，说明是生产者已经接收到消费者已经完全消费了信息，就会删除掉已经消费掉的信息。

    对于生产者宕机，rabbitmq提供了持久化机制，这里的持久化包含了exchange，queue，message的持久化，MessageDeliveryMode的deliveryMode可设置是否持久化，新建exchange和queue的时候也可设置，持久化属性是durable。

2.如何确认消息是否已发送到broker代理服务器上(broker其实就是一个消息队列的服务器实体，包含exchange，queue和binding的信息)

A：方式一:消息队列的channel的confirm模式是针对消息还未到达broker服务器做的一个弥补机制，channel设置成confirm模式后，就可以在到达broker服务器时发送一个反馈(每个消息在发送到broker服务器时都有一个唯一ID)

    方式二:消息队列是基于AMQP协议的，通过AMQP协议的事务机制来实现，是基于AMQP协议层面的解决方案。其实就是Channel中的txSelect()，txCommit()和txRollBack()