Spring整合rabbitmq实践(一):基础
Posted
tags:
篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了Spring整合rabbitmq实践(一):基础相关的知识,希望对你有一定的参考价值。
参考技术A Spring整合rabbitmq实践(二):扩展
Spring整合rabbitmq实践(三):源码
producer:消息生产者;
consumer:消息消费者;
queue:消息队列;
exchange:接收producer发送的消息按照binding规则转发给相应的queue;
binding:exchange与queue之间的关系;
virtualHost:每个virtualHost持有自己的exchange、queue、binding,用户只能在virtualHost粒度控制权限。
fanout:
群发到所有绑定的queue;
direct:
根据routing key routing到相应的queue,routing不到任何queue的消息扔掉;可以不同的key绑到同一个queue,也可以同一个key绑到不同的queue;
topic:
类似direct,区别是routing key是由一组以“.”分隔的单词组成,可以有通配符,“*”匹配一个单词,“#”匹配0个或多个单词;
headers:
根据arguments来routing。
arguments为一组key-value对,任意设置。
“x-match”是一个特殊的key,值为“all”时必须匹配所有argument,值为“any”时只需匹配任意一个argument,不设置默认为“all”。
通过以下配置,可以获得最基础的发送消息到queue,以及从queue接收消息的功能。
这个包同时包含了一些其它的包:spring-context、spring-tx、spring-web、spring-messaging、spring-retry、spring-amqp、amqp-client,如果想单纯一点,可以单独引入。
最主要的是以下几个包,
spring-amqp:
spring-rabbit:
amqp-client:
个人理解就是,spring-amqp是spring整合的amqp,spring-rabbit是spring整合的rabbitmq(rabbitmq是amqp的一个实现,所以可能spring-rabbit也是类似关系),amqp-client提供操作rabbitmq的java api。
目前最新的是2.0.5.RELEASE版本。如果编译报错,以下信息或许能有所帮助:
(1)
解决方案:spring-amqp版本改为2.0.5.RELEASE。
(2)
解决方案:spring-context版本改为5.0.7.RELEASE。
(3)
解决方案:spring-core版本改为5.0.7.RELEASE。
(4)
解决方案:spring-beans版本改为5.0.7.RELEASE。
(5)
解决方案:spring-aop版本改为5.0.7.RELEASE。
总之,需要5.0.7.RELEASE版本的spring,及相匹配版本的amqp-client。
后面所讲的这些bean配置,spring-amqp中都有默认配置,如果不需要修改默认配置,则不用人为配置这些bean。后面这些配置也没有涉及到所有的属性。
这里的ConnectionFactory指的是spring-rabbit包下面的ConnectionFactory接口,不是amqp-client包下面的ConnectionFactory类。
上面这个bean是spring-amqp的核心,不论是发送消息还是接收消息都需要这个bean,下面描述一下里面这些配置的含义。
setAddresses :设置了rabbitmq的地址、端口,集群部署的情况下可填写多个,“,”分隔。
setUsername :设置rabbitmq的用户名。
setPassword :设置rabbitmq的用户密码。
setVirtualHost :设置virtualHost。
setCacheMode :设置缓存模式,共有两种, CHANNEL 和 CONNECTION 模式。
CHANNEL 模式,程序运行期间ConnectionFactory会维护着一个Connection,所有的操作都会使用这个Connection,但一个Connection中可以有多个Channel,操作rabbitmq之前都必须先获取到一个Channel,否则就会阻塞(可以通过setChannelCheckoutTimeout()设置等待时间),这些Channel会被缓存(缓存的数量可以通过setChannelCacheSize()设置);
CONNECTION 模式,这个模式下允许创建多个Connection,会缓存一定数量的Connection,每个Connection中同样会缓存一些Channel,除了可以有多个Connection,其它都跟CHANNEL模式一样。
这里的Connection和Channel是spring-amqp中的概念,并非rabbitmq中的概念,官方文档对Connection和Channel有这样的描述:
关于 CONNECTION 模式中,可以存在多个Connection的使用场景,官方文档的描述:
setChannelCacheSize :设置每个Connection中(注意是每个Connection)可以缓存的Channel数量,注意只是缓存的Channel数量,不是Channel的数量上限,操作rabbitmq之前(send/receive message等)要先获取到一个Channel,获取Channel时会先从缓存中找闲置的Channel,如果没有则创建新的Channel,当Channel数量大于缓存数量时,多出来没法放进缓存的会被关闭。
注意,改变这个值不会影响已经存在的Connection,只影响之后创建的Connection。
setChannelCheckoutTimeout :当这个值大于0时, channelCacheSize 不仅是缓存数量,同时也会变成数量上限,从缓存获取不到可用的Channel时,不会创建新的Channel,会等待这个值设置的毫秒数,到时间仍然获取不到可用的Channel会抛出AmqpTimeoutException异常。
同时,在 CONNECTION 模式,这个值也会影响获取Connection的等待时间,超时获取不到Connection也会抛出AmqpTimeoutException异常。
setPublisherReturns、setPublisherConfirms :producer端的消息确认机制(confirm和return),设为true后开启相应的机制,后文详述。
官方文档描述publisherReturns设为true打开return机制,publisherComfirms设为true打开confirm机制,但测试结果(2.0.5.RELEASE版本)是,任意一个设为true,两个都会打开。
addConnectionListener、addChannelListener、setRecoveryListener :添加或设置相应的Listener,后文详述。
setConnectionCacheSize :仅在 CONNECTION 模式使用,设置Connection的缓存数量。
setConnectionLimit :仅在 CONNECTION 模式使用,设置Connection的数量上限。
上面的bean配置,除了需要注入的几个listener bean以外,其它设置的都是其默认值(2.0.5.RELEASE版本),后面的bean示例配置也是一样,部分属性不同版本的默认值可能有所不同。
一般不用配置这个bean,这里简单提一下。
这个ConnectionFactory是rabbit api中的ConnectionFactory类,这里面是连接rabbitmq节点的Connection配置。
如果想修改这些配置,可以按如下方式配置:
consumer端如果通过@RabbitListener注解的方式接收消息,不需要这个bean。
不建议直接通过ConnectionFactory获取Channel操作rabbitmq,建议通过amqpTemplate操作。
setConnectionFactory :设置spring-amqp的ConnectionFactory。
setRetryTemplate :设置重试机制,详情见后文。
setMessageConverter :设置MessageConverter,用于java对象与Message对象(实际发送和接收的消息对象)之间的相互转换,详情见后文。
setChannelTransacted :打开或关闭Channel的事务,关于amqp的事务后文描述。
setReturnCallback、setConfirmCallback :return和confirm机制的回调接口,后文详述。
setMandatory :设为true使ReturnCallback生效。
这个bean仅在consumer端通过@RabbitListener注解的方式接收消息时使用,每一个@RabbitListener注解的方法都会由这个RabbitListenerContainerFactory创建一个MessageListenerContainer,负责接收消息。
setConnectionFactory :设置spring-amqp的ConnectionFactory。
setMessageConverter :对于consumer端,MessageConverter也可以在这里配置。
setAcknowledgeMode :设置consumer端的应答模式,共有三种:NONE、AUTO、MANUAL。
NONE,无应答,这种模式下rabbitmq默认consumer能正确处理所有发出的消息,所以不管消息有没有被consumer收到,有没有正确处理都不会恢复;
AUTO,由Container自动应答,正确处理发出ack信息,处理失败发出nack信息,rabbitmq发出消息后将会等待consumer端的应答,只有收到ack确认信息才会把消息清除掉,收到nack信息的处理办法由setDefaultRequeueRejected()方法设置,所以在这种模式下,发生错误的消息是可以恢复的。
MANUAL,基本同AUTO模式,区别是需要人为调用方法给应答。
setConcurrentConsumers :设置每个MessageListenerContainer将会创建的Consumer的最小数量,默认是1个。
setMaxConcurrentConsumers :设置每个MessageListenerContainer将会创建的Consumer的最大数量,默认等于最小数量。
setPrefetchCount :设置每次请求发送给每个Consumer的消息数量。
setChannelTransacted :设置Channel的事务。
setTxSize :设置事务当中可以处理的消息数量。
setDefaultRequeueRejected :设置当rabbitmq收到nack/reject确认信息时的处理方式,设为true,扔回queue头部,设为false,丢弃。
setErrorHandler :实现ErrorHandler接口设置进去,所有未catch的异常都会由ErrorHandler处理。
AmqpTamplate里面有下面几个方法可以向queue发送消息:
这里,exchange必须存在,否则消息发不出去,会看到错误日志,但不影响程序运行:
Message是org.springframework.amqp.core.Message类,spring-amqp发送和接收的都是这个Message。
从Message类源码可以看到消息内容放在byte[]里面,MessageProperties对象包含了非常多的一些其它信息,如Header、exchange、routing key等。
这种方式,需要将消息内容(String,或其它Object)转换为byte[],示例:
也可以直接调用下面几个方法,Object将会自动转为Message对象发送:
有两种方法接收消息:
1.polling consumer,轮询调用方法一次获取一条;
2.asynchronous consumer,listener异步接收消息。
polling consumer
直接通过AmqpTemplate的方法从queue获取消息,有如下方法:
如果queue里面没有消息,会立刻返回null;传入timeoutMillis参数后可阻塞等待一段时间。
如果想直接从queue获取想要的java对象,可调用下面这一组方法:
后面4个方法是带泛型的,示例如下:
使用这四个方法需要配置org.springframework.amqp.support.converter.SmartMessageConverter,这是一个接口,Jackson2JsonMessageConverter已经实现了这个接口,所以只要将Jackson2JsonMessageConverter设置到RabbitTemplate中即可。
asynchronous consumer
有多种方式可以实现,详情参考官方文档。
最简单的实现方式是@RabbitListener注解,示例:
这里接收消息的对象用的是Message,也可以是自定义的java对象,但调用Converter转换失败会报错。
注解上指定的queue必须是已经存在并且绑定到某个exchange的,否则会报错:
如果在@RabbitListener注解中指明binding信息,就能自动创建queue、exchange并建立binding关系。
direct和topic类型的exchange需要routingKey,示例:
fanout类型的exchange,示例:
2.0版本之后,可以指定多个routingKey,示例:
并且支持arguments属性,可用于headers类型的exchange,示例:
@Queue有两个参数exclusive和autoDelete顺便解释一下:
exclusive,排他队列,只对创建这个queue的Connection可见,Connection关闭queue删除;
autoDelete,没有consumer对这个queue消费时删除。
对于这两种队列,durable=true是不起作用的。
另外,如果注解申明的queue和exchange及binding关系都已经存在,但与已存在的设置不同,比如,已存在的exchange的是direct类型,这里尝试改为fanout类型,结果是不会有任何影响,不论是修改或者新增参数都不会生效。
如果queue存在,exchange存在,但没有binding,那么程序启动后会自动建立起binding关系。
以上是关于Spring整合rabbitmq实践(一):基础的主要内容,如果未能解决你的问题,请参考以下文章