前言
最近几篇文章将围绕消息中间件RabbitMQ展开,对于RabbitMQ基本概念这里不阐述,主要讲解RabbitMQ的基本用法、Java客户端API介绍、spring Boot与RabbitMQ整合、
Spring Boot与RabbitMQ整合源码分析。
RabbitMQ安装
在使用消息中间件RabbitMQ之前就是安装RabbitMQ。
- 安装erlang:yum install erlang
- 下载RabbitMQ安装包: https://www.rabbitmq.com/releases/rabbitmq-server/v3.5.6/rabbitmq-server-generic-unix-3.5.6.tar.gz
- 解压安装包、配置环境变量RABBITMQ_HOME
RabbitMQ配置
- rabbitmq-env.conf 环境信息配置
rabbitmq.config 核心配置文件
该配置表示是的默认用户guest用户可以远程访问mq(广域网不能访问,内网可以访问)
3. RabbitMQ提供WEB-UI管理控制台,使用 rabbitmq-plugins enable rabbitmq_management命令启用,重启后可以看到
表明WEB-UI控制台启动成功,访问:http://localhost:15672/
创建Test用户
- rabbitmqctl add_user test test
-
rabbitmqctl set_user_tags test administrator
tag分为四种"management", "policymaker", "monitoring" "administrator" 详见 http://www.rabbitmq.com/management.html
RabbitMQ 其他
RabbitMQ Java Client
消息消费者
- 创建连接工厂ConnectionFactory
- 获取连接Connection
- 通过连接获取通信通道Channel
- 声明交换机Exchange:交换机类型分为四类:
FanoutExchange: 将消息分发到所有的绑定队列,无routingkey的概念
HeadersExchange :通过添加属性key-value匹配
DirectExchange:按照routingkey分发到指定队列
TopicExchange:多关键字匹配
声明队列Queue
将队列和交换机绑定
创建消费者
执行消息的消费
消息生产者
- 创建连接工厂ConnectionFactory
- 获取连接Connection
- 通过连接获取通信通道Channel
- 发送消息
package org.lkl.mq.rabbitmq.test; import java.io.IOException;
import java.util.concurrent.TimeoutException; import com.rabbitmq.client.Channel;
import com.rabbitmq.client.ConfirmListener;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.MessageProperties; /**
* 消息publish
*
* @author liaokailin
* @version $Id: Send.java, v 0.1 2015年10月22日 下午3:48:09 liaokailin Exp $
*/
public class Send {
public final static String EXCHANGE_NAME = "test-exchange"; public static void main(String[] args) throws IOException, TimeoutException, InterruptedException {
/**
* 配置amqp broker 连接信息
*/
ConnectionFactory facotry = new ConnectionFactory();
facotry.setUsername("test");
facotry.setPassword("test");
facotry.setVirtualHost("test");
facotry.setHost("localhost"); Connection conn = facotry.newConnection(); //获取一个链接
//通过Channel进行通信
Channel channel = conn.createChannel(); // channel.exchangeDeclare(Send.EXCHANGE_NAME, "direct", true); //如果消费者已创建,这里可不声明
channel.confirmSelect(); //Enables publisher acknowledgements on this channel
channel.addConfirmListener(new ConfirmListener() { @Override
public void handleNack(long deliveryTag, boolean multiple) throws IOException {
System.out.println("[handleNack] :" + deliveryTag + "," + multiple); } @Override
public void handleAck(long deliveryTag, boolean multiple) throws IOException {
System.out.println("[handleAck] :" + deliveryTag + "," + multiple);
}
}); String message = "lkl-";
//消息持久化 MessageProperties.PERSISTENT_TEXT_PLAIN
//发送多条信息,每条消息对应routekey都不一致
for (int i = 0; i < 10; i++) {
channel.basicPublish(EXCHANGE_NAME, message + (i % 2), MessageProperties.PERSISTENT_TEXT_PLAIN,
(message + i).getBytes());
System.out.println("[send] msg " + (message + i) + " of routingKey is " + (message + (i % 2)));
} }
}
否则回调函数无法调用
前言
本篇主要讲述spring Boot与RabbitMQ的整合,内容非常简单,纯API的调用操作。 操作之间需要加入依赖Jar
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-amqp</artifactId>
/dependency>
消息生产者
ConnectionFactory配置
@Configuration
public class AmqpConfig { public static final String EXCHANGE = "spring-boot-exchange";
public static final String ROUTINGKEY = "spring-boot-routingKey"; @Bean
public ConnectionFactory connectionFactory() {
CachingConnectionFactory connectionFactory = new CachingConnectionFactory();
connectionFactory.setAddresses("127.0.0.1:5672");
connectionFactory.setUsername("guest");
connectionFactory.setPassword("guest");
connectionFactory.setVirtualHost("/");
connectionFactory.setPublisherConfirms(true); //必须要设置
return connectionFactory;
}
}
connectionFactory.setPublisherConfirms(true);
RabbitTemplate
通过使用RabbitTemplate来对开发者提供API操作
@Bean
@Scope(ConfigurableBeanFactory.SCOPE_PROTOTYPE)
//必须是prototype类型
public RabbitTemplate rabbitTemplate() {
RabbitTemplate template = new RabbitTemplate(connectionFactory());
return template;
}
public void convertAndSend(String exchange, String routingKey, final Object object, CorrelationData correlationData)
- exchange:交换机名称
routingKey:路由关键字
object:发送的消息内容
correlationData:消息ID
Send.java
@Component
public class Send { private RabbitTemplate rabbitTemplate; /**
* 构造方法注入
*/
@Autowired
public Send(RabbitTemplate rabbitTemplate) {
this.rabbitTemplate = rabbitTemplate;
} public void sendMsg(String content) {
CorrelationData correlationId = new CorrelationData(UUID.randomUUID().toString());
rabbitTemplate.convertAndSend(AmqpConfig.EXCHANGE, AmqpConfig.ROUTINGKEY, content, correlationId);
} }
如果需要在生产者需要消息发送后的回调,需要对rabbitTemplate设置ConfirmCallback对象,由于不同的生产者需要对应不同的ConfirmCallback,如果rabbitTemplate设置为单例bean,则所有的rabbitTemplate
实际的ConfirmCallback为最后一次申明的ConfirmCallback。
下面给出完整的生产者代码:
package com.lkl.springboot.amqp; 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.stereotype.Component; /**
* 消息生产者
*
* @author liaokailin
* @version $Id: Send.java, v 0.1 2015年11月01日 下午4:22:25 liaokailin Exp $
*/
@Component
public class Send implements RabbitTemplate.ConfirmCallback { private RabbitTemplate rabbitTemplate; /**
* 构造方法注入
*/
@Autowired
public Send(RabbitTemplate rabbitTemplate) {
this.rabbitTemplate = rabbitTemplate;
rabbitTemplate.setConfirmCallback(this); //rabbitTemplate如果为单例的话,那回调就是最后设置的内容
} public void sendMsg(String content) {
CorrelationData correlationId = new CorrelationData(UUID.randomUUID().toString());
rabbitTemplate.convertAndSend(AmqpConfig.EXCHANGE, AmqpConfig.ROUTINGKEY, content, correlationId);
} /**
* 回调
*/
@Override
public void confirm(CorrelationData correlationData, boolean ack, String cause) {
System.out.println(" 回调id:" + correlationData);
if (ack) {
System.out.println("消息成功消费");
} else {
System.out.println("消息消费失败:" + cause);
}
} }
消息消费者
消费者负责申明交换机(生产者也可以申明)、队列、两者的绑定操作。
交换机
/**
* 针对消费者配置
FanoutExchange: 将消息分发到所有的绑定队列,无routingkey的概念
HeadersExchange :通过添加属性key-value匹配
DirectExchange:按照routingkey分发到指定队列
TopicExchange:多关键字匹配
*/
@Bean
public DirectExchange defaultExchange() {
return new DirectExchange(EXCHANGE);
}
在Spring Boot中交换机继承AbstractExchange类
队列
@Bean
public Queue queue() {
return new Queue("spring-boot-queue", true); //队列持久 }
绑定
消息消费
@Bean
public SimpleMessageListenerContainer messageContainer() {
SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory());
container.setQueues(queue());
container.setExposeListenerChannel(true);
container.setMaxConcurrentConsumers(1);
container.setConcurrentConsumers(1);
container.setAcknowledgeMode(AcknowledgeMode.MANUAL); //设置确认模式手工确认
container.setMessageListener(new ChannelAwareMessageListener() { @Override
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;
}
下面给出完整的配置文件:
package com.lkl.springboot.amqp; 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.Message;
import org.springframework.amqp.core.Queue;
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.beans.factory.config.ConfigurableBeanFactory;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Scope; import com.rabbitmq.client.Channel; /**
* Qmqp Rabbitmq
*
* http://docs.spring.io/spring-amqp/docs/1.4.5.RELEASE/reference/html/
*
* @author lkl
* @version $Id: AmqpConfig.java, v 0.1 2015年11月01日 下午2:05:37 lkl Exp $
*/ @Configuration
public class AmqpConfig { public static final String EXCHANGE = "spring-boot-exchange";
public static final String ROUTINGKEY = "spring-boot-routingKey"; @Bean
public ConnectionFactory connectionFactory() {
CachingConnectionFactory connectionFactory = new CachingConnectionFactory();
connectionFactory.setAddresses("127.0.0.1:5672");
connectionFactory.setUsername("guest");
connectionFactory.setPassword("guest");
connectionFactory.setVirtualHost("/");
connectionFactory.setPublisherConfirms(true); //必须要设置
return connectionFactory;
} @Bean
@Scope(ConfigurableBeanFactory.SCOPE_PROTOTYPE)
//必须是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(EXCHANGE);
} @Bean
public Queue queue() {
return new Queue("spring-boot-queue", true); //队列持久 } @Bean
public Binding binding() {
return BindingBuilder.bind(queue()).to(defaultExchange()).with(AmqpConfig.ROUTINGKEY);
} @Bean
public SimpleMessageListenerContainer messageContainer() {
SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory());
container.setQueues(queue());
container.setExposeListenerChannel(true);
container.setMaxConcurrentConsumers(1);
container.setConcurrentConsumers(1);
container.setAcknowledgeMode(AcknowledgeMode.MANUAL); //设置确认模式手工确认
container.setMessageListener(new ChannelAwareMessageListener() { @Override
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;
} }
自动配置
spring.rabbitmq.host=localhost
spring.rabbitmq.port=5672
spring.rabbitmq.username=test
spring.rabbitmq.password=test
spring.rabbitmq.virtualHost=test
connectionFactory.setPublisherConfirms(true);
=========================================
前言
RabbitAdmin
在上篇中遗留AmqpAdmin没有讲解,现在来看下该部分代码
创建RabbitAdmin实例,调用构造方法
创建连接工厂、rabbitTemplate,其中ConnectionFactory采用上一篇中自定义bean
为CachingConnectionFactory实例,其缓存模式为通道缓存
private volatile CacheMode cacheMode = CacheMode.CHANNEL;
public class RabbitAdmin implements AmqpAdmin, ApplicationContextAware, InitializingBean {
...
}
实现接口AmqpAdmin(定义若干RabbitMQ操作父接口),这里需要强调的是InitializingBean,实现该接口则会调用afterPropertiesSet方法
public void afterPropertiesSet() { synchronized (this.lifecycleMonitor) { if (this.running || !this.autoStartup) {
return;
} if (this.connectionFactory instanceof CachingConnectionFactory &&
((CachingConnectionFactory) this.connectionFactory).getCacheMode() == CacheMode.CONNECTION) {
logger.warn("RabbitAdmin auto declaration is not supported with CacheMode.CONNECTION");
return;
} this.connectionFactory.addConnectionListener(new ConnectionListener() { // Prevent stack overflow...
private final AtomicBoolean initializing = new AtomicBoolean(false); @Override
public void onCreate(Connection connection) {
if (!initializing.compareAndSet(false, true)) {
// If we are already initializing, we don't need to do it again...
return;
}
try { initialize();
}
finally {
initializing.compareAndSet(true, false);
}
} @Override
public void onClose(Connection connection) {
} }); this.running = true; }
}
synchronized (this.lifecycleMonitor)加锁保证同一时间只有一个线程访问该代码,随后调用this.connectionFactory.addConnectionListener添加连接监听,各连接工厂关系:
此时connection为null,无法执行到listener.onCreate(this.connection);
往CompositeConnectionListener connectionListener中添加监听信息,最终保证在集合中
Exchange
以上代码创建一个交换机,交换机类型为direct
Queue
Binding
public static DestinationConfigurer bind(Queue queue) {
return new DestinationConfigurer(queue.getName(), DestinationType.QUEUE);
}
DestinationConfigurer通过name、type区分不同配置信息,其to()方法为重载方法,传递参数为四种交换机,分别返回XxxExchangeRoutingKeyConfigurer,其中with方法返回Bingding实例,因此在Binding信息中存储了
以上信息理解都非常简单,下面来看比较复杂点的SimpleMessageListenerContainer
SimpleMessageListenerContainer
查看其实现的接口,注意SmartLifecycle
private volatile List<String> queueNames = new CopyOnWriteArrayList<String>();
添加队列信息
设置并发消费者数量,默认情况为1
container.setAcknowledgeMode(AcknowledgeMode.MANUAL); //设置确认模式手工确认
设置消费者成功消费消息后确认模式,分为两种
- 自动模式,默认模式,在RabbitMQ Broker消息发送到消费者后自动删除
- 手动模式,消费者客户端显示编码确认消息消费完成,Broker给生产者发送回调,消息删除
SmartLifecycle
其中的isAutoStartup设置为true时,会自动调用Lifecycle接口中的start方法,既然我们为源码分析,也简单看下这个聪明的声明周期接口是如何实现它的聪明方法的
- protected void finishRefresh() {
- // Initialize lifecycle processor for this context.
- initLifecycleProcessor();
- // Propagate refresh to lifecycle processor first.
- getLifecycleProcessor().onRefresh();
- // Publish the final event.
- publishEvent(new ContextRefreshedEvent(this));
- // Participate in LiveBeansView MBean, if active.
- LiveBeansView.registerApplicationContext(this);
- }
其中initLifecycleProcessor初始化生命周期处理器,
- protected void initLifecycleProcessor() {
- ConfigurableListableBeanFactory beanFactory = getBeanFactory();
- if (beanFactory.containsLocalBean(LIFECYCLE_PROCESSOR_BEAN_NAME)) {
- this.lifecycleProcessor =
- beanFactory.getBean(LIFECYCLE_PROCESSOR_BEAN_NAME, LifecycleProcessor.class);
- if (logger.isDebugEnabled()) {
- logger.debug("Using LifecycleProcessor [" + this.lifecycleProcessor + "]");
- }
- }
- else {
- DefaultLifecycleProcessor defaultProcessor = new DefaultLifecycleProcessor();
- defaultProcessor.setBeanFactory(beanFactory);
- this.lifecycleProcessor = defaultProcessor;
- beanFactory.registerSingleton(LIFECYCLE_PROCESSOR_BEAN_NAME, this.lifecycleProcessor);
- if (logger.isDebugEnabled()) {
- logger.debug("Unable to locate LifecycleProcessor with name '" +
- LIFECYCLE_PROCESSOR_BEAN_NAME +
- "': using default [" + this.lifecycleProcessor + "]");
- }
- }
- }
注册DefaultLifecycleProcessor对应bean
getLifecycleProcessor().onRefresh()调用DefaultLifecycleProcessor中方法onRefresh,调用startBeans(true)
- private void startBeans(boolean autoStartupOnly) {
- Map<String, Lifecycle> lifecycleBeans = getLifecycleBeans();
- Map<Integer, LifecycleGroup> phases = new HashMap<Integer, LifecycleGroup>();
- for (Map.Entry<String, ? extends Lifecycle> entry : lifecycleBeans.entrySet()) {
- Lifecycle bean = entry.getValue();
- if (!autoStartupOnly || (bean instanceof SmartLifecycle && ((SmartLifecycle) bean).isAutoStartup())) {
- int phase = getPhase(bean);
- LifecycleGroup group = phases.get(phase);
- if (group == null) {
- group = new LifecycleGroup(phase, this.timeoutPerShutdownPhase, lifecycleBeans, autoStartupOnly);
- phases.put(phase, group);
- }
- group.add(entry.getKey(), bean);
- }
- }
- if (phases.size() > 0) {
- List<Integer> keys = new ArrayList<Integer>(phases.keySet());
- Collections.sort(keys);
- for (Integer key : keys) {
- phases.get(key).start();
- }
- }
- }
其中
Map<String, Lifecycle> lifecycleBeans = getLifecycleBeans();
获取所有实现Lifecycle接口bean,执行bean instanceof SmartLifecycle && ((SmartLifecycle)bean).isAutoStartup()判断,如果bean同时也为Phased实例,则加入到LifecycleGroup中,随后phases.get(key).start()调用start方法
接下来要做的事情就很明显:要了解消费者具体如何实现,查看SimpleMessageListenerContainer中的start是如何实现的。
至此~~整合RabbitMQ源码分析准备工作完成,下一篇中正式解读消费者的实现。
==============================
踩坑记录
近日在用spring boot架构一个微服务框架,服务发现与治理、发布REST接口各种轻松惬意。但是服务当设计MQ入口时,就发现遇到无数地雷,现在整理成下文,供各路大侠围观与嘲笑。
版本
当前使用的spring-boot-starter-amqp版本为2016.5发布的1.3.5.RELEASE
也许若干年后,你们版本都不会有这些问题了。:(
RabbitMQ
当需要用到MQ的时候,我的第一反映就是使用RabbitMQ,猫了一眼spring boot的官方说明,上面说spring boot为rabbit准备了spring-boot-starter-amqp,并且为RabbitTemplate和RabbitMQ提供了自动配置选项。暗自窃喜~~
瞅瞅[官方文档]http://docs.spring.io/spring-boot/docs/current-SNAPSHOT/reference/htmlsingle/#boot-features-rabbitmq和例子,SO EASY,再看一眼GITHUB上的官方例了,也有例子。
心情愉悦的照着例子,开干~~。
踩坑
十五分钟后的代码类似这样:
@Service
@RabbitListener(queues = "merchant")
public class MQReceiver {
protected Logger logger = Logger.getLogger(MQReceiver.class
.getName());
@RabbitHandler
public void process(@Payload UpdateMerchant request) {
UpdateMerchantResponse response = new UpdateMerchantResponse();
logger.info(request.getMerchantId() + "->" + response.getReturnCode());
}
}
消费信息后,应该记录一条日志。
结果得到只有org.springframework.amqp.AmqpException: No method found for class [B 这个异常,并且还无限循环抛出这个异常。。。
记得刚才官方文档好像说了异常什么的,转身去猫一眼,果然有:
If retries are not enabled and the listener throws an exception, by default the delivery will be retried indefinitely. You can modify this behavior in two ways; set the defaultRequeueRejected
property to false
and zero re-deliveries will be attempted; or, throw an AmqpRejectAndDontRequeueException
to signal the message should be rejected. This is the mechanism used when retries are enabled and the maximum delivery attempts are reached.
知道了为啥会无限重试了,下面来看看为啥会抛出这个异常,google搜一下,貌似还有一个倒霉鬼遇到了这个问题。
进去看完问题和大神的解答,豁然开朗。
There are two conversions in the @RabbitListener pipeline.
The first converts from a Spring AMQP Message to a spring-messaging Message.
There is currently no way to change the first converter from SimpleMessageConverter which handles String, Serializable and passes everything else as byte[].
The second converter converts the message payload to the method parameter type (if necessary).
With method-level @RabbitListeners there is a tight binding between the handler and the method.
With class-level @RabbitListener s, the message payload from the first conversion is used to select which method to invoke. Only then, is the argument conversion attempted.
This mechanism works fine with Java Serializable objects since the payload has already been converted before the method is selected.
However, with JSON, the first conversion returns a byte[] and hence we find no matching @RabbitHandler.
We need a mechanism such that the first converter is settable so that the payload is converted early enough in the pipeline to select the appropriate handler method.
A ContentTypeDelegatingMessageConverter is probably most appropriate.
And, as stated in AMQP-574, we need to clearly document the conversion needs for a @RabbitListener, especially when using JSON or a custom conversion.
得嘞,官方示例果然是坑,试试大神的解决方案,手动新增下转换。
@Bean
public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) {
RabbitTemplate template = new RabbitTemplate(connectionFactory);
template.setMessageConverter(new Jackson2JsonMessageConverter());
return template;
}
@Bean
public SimpleRabbitListenerContainerFactory rabbitListenerContainerFactory(ConnectionFactory connectionFactory) {
SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
factory.setConnectionFactory(connectionFactory);
factory.setMessageConverter(new Jackson2JsonMessageConverter());
return factory;
}
然后在生产和消费信息的地方使用他们:
@RabbitListener(queues = "merchant", containerFactory="rabbitListenerContainerFactory")
public void process(@Payload UpdateMerchant request) {
UpdateMerchantResponse response = new UpdateMerchantResponse();
logger.info(request.getMerchantId() + "->" + response.getReturnCode());
}
再来一次,果然可以了
c.l.s.m.service.MQReceiver : 00000001->null