- channel.exchangeDeclare()
- chanel.basicQos()
- channel.basicPublish()
- channel.basicAck()
- channel.basicNack(delivery.getEnvelope().getDeliveryTag(), false, true)
- channel.basicReject(delivery.getEnvelope().getDeliveryTag(), false)
- channel.basicConsume(queueName, true, consumer)
- chanel.exchangeBind()
- channel.queueDeclare(queueName, false, false, false, null)
channel.exchangeDeclare()
type
:有direct
、fanout
、topic
三种durable
:(true、false
) true:服务器重启会保留下来Exchange。警告:仅设置此选项,不代表消息持久化。即不保证重启后消息还在
原文:true if we are declaring a durable exchange (the exchange will survive a server restart)
- autoDelete:(
true、false
) true:当已经没有消费者时,服务器是否可以删除该Exchange
原文1:true if the server should delete the exchange when it is no longer in use
/**
* Declare an exchange.
* @see com.rabbitmq.client.AMQP.Exchange.Declare
* @see com.rabbitmq.client.AMQP.Exchange.DeclareOk
* @param exchange the name of the exchange
* @param type the exchange type
* @param durable true if we are declaring a durable exchange (the exchange will survive a server restart)
* @param autoDelete true if the server should delete the exchange when it is no longer in use
* @param arguments other properties (construction arguments) for the exchange
* @return a declaration-confirm method to indicate the exchange was successfully declared
* @throws java.io.IOException if an error is encountered
*/
Exchange.DeclareOk exchangeDeclare(String exchange, String type, boolean durable, boolean autoDelete,
Map<String, Object> arguments) throws IOException;
chanel.basicQos()
prefetchSize
:0prefetchCount
:会告诉RabbitMQ不要同时给一个消费者推送多于N个消息,即一旦有N个消息还没有ack,则该consumer将block掉,直到有消息ackglobal
:true\false
是否将上面设置应用于channel,简单点说,就是上面限制是channel
级别的还是consumer
级别
备注:据说prefetchSize 和global这两项,rabbitmq没有实现,暂且不研究
/**
* Request specific "quality of service" settings.
*
* These settings impose limits on the amount of data the server
* will deliver to consumers before requiring acknowledgements.
* Thus they provide a means of consumer-initiated flow control.
* @see com.rabbitmq.client.AMQP.Basic.Qos
* @param prefetchSize maximum amount of content (measured in
* octets) that the server will deliver, 0 if unlimited
* @param prefetchCount maximum number of messages that the server
* will deliver, 0 if unlimited
* @param global true if the settings should be applied to the
* entire channel rather than each consumer
* @throws java.io.IOException if an error is encountered
*/
void basicQos(int prefetchSize, int prefetchCount, boolean global) throws IOException;
channel.basicPublish()
routingKey
:路由键,#匹配0个或多个单词,*匹配一个单词,在topic exchange做消息转发用mandatory
:true:如果exchange根据自身类型和消息routeKey无法找到一个符合条件的queue,那么会调用basic.return方法将消息返还给生产者。false:出现上述情形broker会直接将消息扔掉immediate
:true:如果exchange在将消息route到queue(s)时发现对应的queue上没有消费者,那么这条消息不会放入队列中。当与消息routeKey关联的所有queue(一个或多个)都没有消费者时,该消息会通过basic.return方法返还给生产者。BasicProperties
:需要注意的是BasicProperties.deliveryMode
,0:不持久化 1:持久化 这里指的是消息的持久化,配合channel(durable=true),queue(durable)可以实现,即使服务器宕机,消息仍然保留
简单来说:
mandatory
标志告诉服务器至少将该消息route到一个队列中,否则将消息返还给生产者;immediate
标志告诉服务器如果该消息关联的queue上有消费者,则马上将消息投递给它,如果所有queue都没有消费者,直接把消息返还给生产者,不用将消息入队列等待消费者了
/**
* Publish a message.
*
* Publishing to a non-existent exchange will result in a channel-level
* protocol exception, which closes the channel.
*
* Invocations of <code>Channel#basicPublish</code> will eventually block if a
* <a href="http://www.rabbitmq.com/alarms.html">resource-driven alarm</a> is in effect.
*
* @see com.rabbitmq.client.AMQP.Basic.Publish
* @see <a href="http://www.rabbitmq.com/alarms.html">Resource-driven alarms</a>.
* @param exchange the exchange to publish the message to
* @param routingKey the routing key
* @param mandatory true if the 'mandatory' flag is to be set
* @param immediate true if the 'immediate' flag is to be
* set. Note that the RabbitMQ server does not support this flag.
* @param props other properties for the message - routing headers etc
* @param body the message body
* @throws java.io.IOException if an error is encountered
*/
void basicPublish(String exchange, String routingKey, boolean mandatory, boolean immediate, BasicProperties props, byte[] body)
throws IOException;
channel.basicAck()
deliveryTag
:该消息的indexmultiple
:是否批量;true:将一次性ack所有小于deliveryTag的消息
/**
* Acknowledge one or several received
* messages. Supply the deliveryTag from the {@link com.rabbitmq.client.AMQP.Basic.GetOk}
* or {@link com.rabbitmq.client.AMQP.Basic.Deliver} method
* containing the received message being acknowledged.
* @see com.rabbitmq.client.AMQP.Basic.Ack
* @param deliveryTag the tag from the received {@link com.rabbitmq.client.AMQP.Basic.GetOk} or {@link com.rabbitmq.client.AMQP.Basic.Deliver}
* @param multiple true to acknowledge all messages up to and
* including the supplied delivery tag; false to acknowledge just
* the supplied delivery tag.
* @throws java.io.IOException if an error is encountered
*/
void basicAck(long deliveryTag, boolean multiple) throws IOException;
channel.basicNack(delivery.getEnvelope().getDeliveryTag(), false, true)
deliveryTag
:该消息的indexmultiple
:是否批量.true:将一次性拒绝所有小于deliveryTag的消息requeue
:被拒绝的是否重新入队列
/**
* Reject one or several received messages.
*
* Supply the <code>deliveryTag</code> from the {@link com.rabbitmq.client.AMQP.Basic.GetOk}
* or {@link com.rabbitmq.client.AMQP.Basic.GetOk} method containing the message to be rejected.
* @see com.rabbitmq.client.AMQP.Basic.Nack
* @param deliveryTag the tag from the received {@link com.rabbitmq.client.AMQP.Basic.GetOk} or {@link com.rabbitmq.client.AMQP.Basic.Deliver}
* @param multiple true to reject all messages up to and including
* the supplied delivery tag; false to reject just the supplied
* delivery tag.
* @param requeue true if the rejected message(s) should be requeued rather
* than discarded/dead-lettered
* @throws java.io.IOException if an error is encountered
*/
void basicNack(long deliveryTag, boolean multiple, boolean requeue)
throws IOException;
channel.basicReject(delivery.getEnvelope().getDeliveryTag(), false)
deliveryTag
:该消息的indexrequeue
:被拒绝的是否重新入队列
channel.basicNack 与 channel.basicReject 的区别在于basicNack可以拒绝多条消息,而basicReject一次只能拒绝一条消息
/**
* Reject a message. Supply the deliveryTag from the {@link com.rabbitmq.client.AMQP.Basic.GetOk}
* or {@link com.rabbitmq.client.AMQP.Basic.Deliver} method
* containing the received message being rejected.
* @see com.rabbitmq.client.AMQP.Basic.Reject
* @param deliveryTag the tag from the received {@link com.rabbitmq.client.AMQP.Basic.GetOk} or {@link com.rabbitmq.client.AMQP.Basic.Deliver}
* @param requeue true if the rejected message should be requeued rather than discarded/dead-lettered
* @throws java.io.IOException if an error is encountered
*/
void basicReject(long deliveryTag, boolean requeue) throws IOException;
channel.basicConsume(queueName, true, consumer)
autoAck
:是否自动ack
,如果不自动ack
,需要使用channel.ack
、channel.nack
、channel.basicReject
进行消息应答
/**
* Start a non-nolocal, non-exclusive consumer, with
* a server-generated consumerTag.
* @param queue the name of the queue
* @param autoAck true if the server should consider messages
* acknowledged once delivered; false if the server should expect
* explicit acknowledgements
* @param callback an interface to the consumer object
* @return the consumerTag generated by the server
* @throws java.io.IOException if an error is encountered
* @see com.rabbitmq.client.AMQP.Basic.Consume
* @see com.rabbitmq.client.AMQP.Basic.ConsumeOk
* @see #basicConsume(String, boolean, String, boolean, boolean, Map, Consumer)
*/
String basicConsume(String queue, boolean autoAck, Consumer callback) throws IOException;
chanel.exchangeBind()
channel.queueBind(queueName, EXCHANGE_NAME, bindingKey)
;用于通过绑定bindingKey
将queue
到Exchange
,之后便可以进行消息接收
/**
* Bind an exchange to an exchange, with no extra arguments.
* @see com.rabbitmq.client.AMQP.Exchange.Bind
* @see com.rabbitmq.client.AMQP.Exchange.BindOk
* @param destination the name of the exchange to which messages flow across the binding
* @param source the name of the exchange from which messages flow across the binding
* @param routingKey the routine key to use for the binding
* @return a binding-confirm method if the binding was successfully created
* @throws java.io.IOException if an error is encountered
*/
Exchange.BindOk exchangeBind(String destination, String source, String routingKey) throws IOException;
channel.queueDeclare(queueName, false, false, false, null)
durable
:(true、false) true:在服务器重启时,能够存活exclusive
:是否为当前连接的专用队列,在连接断开后,会自动删除该队列,生产环境中应该很少用到autodelete
:当没有任何消费者使用时,自动删除该队列
this means that the queue will be deleted when there are no more processes consuming messages from it.)
/**
* Declare a queue
* @see com.rabbitmq.client.AMQP.Queue.Declare
* @see com.rabbitmq.client.AMQP.Queue.DeclareOk
* @param queue the name of the queue
* @param durable true if we are declaring a durable queue (the queue will survive a server restart)
* @param exclusive true if we are declaring an exclusive queue (restricted to this connection)
* @param autoDelete true if we are declaring an autodelete queue (server will delete it when no longer in use)
* @param arguments other properties (construction arguments) for the queue
* @return a declaration-confirm method to indicate the queue was successfully declared
* @throws java.io.IOException if an error is encountered
*/
Queue.DeclareOk queueDeclare(String queue, boolean durable, boolean exclusive, boolean autoDelete,
Map<String, Object> arguments) throws IOException;
arguments
中可以设置的常用属性
参数名 | 作用 | 示例 | 示例描述 |
---|---|---|---|
Message TTL | 设置队列中消息的有效时间 | { “x-message-ttl”,1000*8} | 设置队列中的所有消息的有效期为8s |
Auto expire | 一定的时间内队列没有被使用,则自动删除队列 | {“x-expires”,1000*60} | 如果60s没有队列被访问,则删除队列 |
Max length | 队列能保存消息的最大条数 | {“x-max-length”,100 } | 设置队列最多保存100条消息 |
Max length bytes | 队列中ready类型消息的总字节数 | {“x-max-length-bytes”,1000 } | 设置队列中ready类型消息总共不能超过1000字节 |
Overflow behaviour | 当队列消息满了时,再接收消息时的处理方法。有两种处理方案:默认为”drop-head”模式,表示从队列头部丢弃消息;”reject-publish”表示不接收后续的消息 | {“x-overflow”,”reject-publish” } | 设置当队列消息满了时,丢弃传来后续消息 |
Dead letter exchange | 用于存储被丢弃的消息的交换机名。Overflow behaviour 的两种处理方案中丢弃的消息都会发送到这个交换机 | {“x-dead-letter-exchange”,”beiyongExchange” } | 设置丢弃的消息发送到名字为beiyongExchange的交换机 |
Dead letter routing key | 被丢弃的消息发送到Dead letter exchange时的使用的routing Key | {“x-dead-letter-routing-key”,”deadKey” } | 设置丢弃的消息发送到beiyongExchange交换机时的RoutingKey值是”deadKey” |
Maximum priority | 设置队列中消息优先级的最大等级,在publish消息时可以设置单条消息的优先级等级 | {“x-max-priority”,10 } | 设置中消息优先级的最大等级为10 |
Lazy mode | 设置队列的模式,如果设置为Lazy表示队列中消息尽可能存放在磁盘中,以减少内存占用;不设置时消息都存放在队列中,用以尽可能快的处理消息 | {“x-queue-mode”,”lazy”} | 3.6以后版本可用,设置队列中消息尽可能存放在磁盘中,以减少内存占用。在消息拥堵时和消息持久化配置使用可以减少内存占用 |