java API编程

简单实例

引入pom

<dependency><groupId>com.rabbitmq</groupId><artifactId>amqp-client</artifactId><version>4.1.0</version></dependency>

定义生产者

package com.example.rabbitmq.producer;import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;import java.io.IOException;public class rbqProducer {private final static String EXCHANGE_NAME = "SIMPLE_EXCHANGE";public static void main(String[] args) throws Exception {ConnectionFactory factory = new ConnectionFactory();// 连接IPfactory.setHost("192.168.40.129");// 连接端口factory.setPort(5672);// 虚拟机factory.setVirtualHost("/");// 用户factory.setUsername("guest");factory.setPassword("guest");// 建立连接Connection conn = factory.newConnection();// 创建消息通道Channel channel = conn.createChannel();// 发送消息String msg = "Hello world, Rabbit MQ";// String exchange, String routingKey, BasicProperties props, byte[] bodychannel.basicPublish(EXCHANGE_NAME, "lulf.test", null, msg.getBytes());channel.close();conn.close();}
}

定义消费者

package com.example.rabbitmq.consumer;import com.rabbitmq.client.*;import java.io.IOException;public class rbqConsumer {private final static String EXCHANGE_NAME = "SIMPLE_EXCHANGE";private final static String QUEUE_NAME = "SIMPLE_QUEUE";public static void main(String[] args) throws Exception {ConnectionFactory factory = new ConnectionFactory();// 连接IPfactory.setHost("192.168.40.129");// 默认监听端口factory.setPort(5672);// 虚拟机factory.setVirtualHost("/");// 设置访问的用户factory.setUsername("guest");factory.setPassword("guest");// 建立连接Connection conn = factory.newConnection();// 创建消息通道Channel channel = conn.createChannel();// 声明交换机// String exchange, String type, boolean durable, boolean autoDelete, Map<String, Object> argumentschannel.exchangeDeclare(EXCHANGE_NAME,"direct",false, false, null);// 声明队列// String queue, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> argumentschannel.queueDeclare(QUEUE_NAME, false, false, false, null);System.out.println(" Waiting for message....");// 绑定队列和交换机channel.queueBind(QUEUE_NAME,EXCHANGE_NAME,"lulf.test");// 创建消费者Consumer consumer = new DefaultConsumer(channel) {@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties,byte[] body) throws IOException {String msg = new String(body, "UTF-8");System.out.println("Received message : '" + msg + "'");System.out.println("consumerTag : " + consumerTag );System.out.println("deliveryTag : " + envelope.getDeliveryTag() );}};// 开始获取消息// String queue, boolean autoAck, Consumer callbackchannel.basicConsume(QUEUE_NAME, true, consumer);}}

amqp client 方法参数说明

声明交换机的参数：
String type：交换机的类型，direct, topic, fanout中的一种。

boolean durable：是否持久化，代表交换机在服务器重启后是否还存在。

boolean autoDelete：是否自动删除。

声明队列的参数：
boolean durable：是否持久化，代表队列在服务器重启后是否还存在。

boolean exclusive：是否排他性队列。排他性队列只能在声明它的Connection中使用，连接断开时自动删除。

boolean autoDelete：是否自动删除。如果为true，至少有一个消费者连接到这个队列，之后所有与这个队列连接的消费者都断开时，队列会自动删除。

Map<String, Object> arguments：队列的其他属性，例如x-message-ttl、x-expires、x-max-length、x-maxlength-bytes、x-dead-letter-exchange、x-dead-letter-routing-key、x-max-priority。

消息属性BasicProperties：

消息的全部属性有14个，以下列举了一些主要的参数：

Map<String,Object> headers 消息的其他自定义参数

Integer deliveryMode 2持久化，其他：瞬态

Integer priority 消息的优先级

String correlationId 关联ID，方便RPC相应与请求关联

String replyTo 回调队列

String expiration TTL，消息过期时间，单位毫秒

RabbitMQ 常见配置

TTL（Time To Live）消息的过期时间

有两种设置方式：通过队列属性设置消息过期时间，设置单条消息的过期时间。
通过队列设置消息过期时间

// 通过队列属性设置消息过期时间
Map<String, Object> argss = new HashMap<String, Object>();
argss.put("x-message-ttl",6000);// 声明队列（默认交换机AMQP default，Direct）
// String queue, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments
channel.queueDeclare("TEST_TTL_QUEUE", false, false, false, argss);

设置单条消息的过期时间

// 对每条消息设置过期时间
AMQP.BasicProperties properties = new AMQP.BasicProperties.Builder().deliveryMode(2) // 持久化消息.contentEncoding("UTF-8").expiration("10000") // TTL.build();// 发送消息
channel.basicPublish("", "TEST_DLX_QUEUE", properties, msg.getBytes());

注意：
如果两种方式设置消息过期时间的方式都使用了，将以较小的数值为准

死信队列

死信，在官网中对应的单词为“Dead Letter”，可以看出翻译确实非常的简单粗暴。那么死信是个什么东西呢？

“死信”是RabbitMQ中的一种消息机制，当你在消费消息时，如果队列里的消息出现以下情况：

消息被否定确认，使用 channel.basicNack 或 channel.basicReject ，并且此时requeue 属性被设置为false。
消息在队列的存活时间超过设置的TTL时间。
消息队列的消息数量已经超过最大队列长度。
那么该消息将成为“死信”。

“死信”消息会被RabbitMQ进行特殊处理，如果配置了死信队列信息，那么该消息将会被丢进死信队列中，如果没有配置，则该消息将会被丢弃

死信队列

package com.example.rabbitmq.consumer;import com.rabbitmq.client.*;import java.io.IOException;
import java.util.HashMap;
import java.util.Map;public class rbqConsumer {private final static String EXCHANGE_NAME = "SIMPLE_EXCHANGE";private final static String QUEUE_NAME = "SIMPLE_QUEUE";public static void main(String[] args) throws Exception {ConnectionFactory factory = new ConnectionFactory();// 连接IPfactory.setHost("192.168.40.129");// 默认监听端口factory.setPort(5672);// 虚拟机factory.setVirtualHost("/");// 设置访问的用户factory.setUsername("guest");factory.setPassword("guest");// 建立连接Connection conn = factory.newConnection();// 创建消息通道Channel channel = conn.createChannel();Map<String,Object> arguments = new HashMap<String,Object>();arguments.put("x-dead-letter-exchange","DLX_EXCHANGE");// 指定了这个队列的死信交换机channel.queueDeclare("TEST_DLX_QUEUE", false, false, false, arguments);// 声明死信交换机channel.exchangeDeclare("DLX_EXCHANGE","topic", false, false, false, null);// 声明死信队列channel.queueDeclare("DLX_QUEUE", false, false, false, null);// 绑定channel.queueBind("DLX_QUEUE","DLX_EXCHANGE","#");// 创建消费者Consumer consumer = new DefaultConsumer(channel) {@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties,byte[] body) throws IOException {String msg = new String(body, "UTF-8");System.out.println("Received message : '" + msg + "'");if (msg.contains("拒收")) {// 拒绝消息// requeue：是否重新入队列，true：是；false：直接丢弃，相当于告诉队列可以直接删除掉// TODO 如果只有这一个消费者，requeue 为true 的时候会造成消息重复消费channel.basicReject(envelope.getDeliveryTag(), false);} else if (msg.contains("异常")) {// 批量拒绝// requeue：是否重新入队列// TODO 如果只有这一个消费者，requeue 为true 的时候会造成消息重复消费channel.basicNack(envelope.getDeliveryTag(), true, false);} else {// 手工应答// 如果不应答，队列中的消息会一直存在，重新连接的时候会重复消费channel.basicAck(envelope.getDeliveryTag(), true);}}};// 开始获取消息// String queue, boolean autoAck, Consumer callbackchannel.basicConsume(QUEUE_NAME, true, consumer);}}

优先队列

可以通过创建队列的时候设置一个队列的优先级的最大值，然后设置指定消息的优先级的值。优先级高的消息可以优先被消费，但是：只有消息堆积（消息的发送速度大于消费者的消费速度）的情况下优先级才有意义。

Map<String, Object> argss = new HashMap<String, Object>();
argss.put("x-max-priority",10); // 队列最大优先级
channel.queueDeclare("ORIGIN_QUEUE", false, false, false, argss);Map<String, Object> headers = new HashMap<String, Object>();headers.put("name", "lulf");headers.put("level", "top");AMQP.BasicProperties properties = new AMQP.BasicProperties.Builder().deliveryMode(2)   // 2代表持久化.contentEncoding("UTF-8")  // 编码.expiration("10000")  // TTL，过期时间.headers(headers) // 自定义属性.priority(5) // 优先级，默认为5，配合队列的 x-max-priority 属性使用.messageId(String.valueOf(UUID.randomUUID())).build();
channel.basicPublish("", "ORIGIN_QUEUE", properties, msg.getBytes());

延迟队列

RabbitMQ本身不支持延迟队列。可以使用TTL结合DLX的方式来实现消息的延迟投递，即把DLX跟某个队列绑定，到了指定时间，消息过期后，就会从DLX路由到这个队列，消费者可以从这个队列取走消息。

另一种方式是使用rabbitmq-delayed-message-exchange插件。当然，将需要发送的信息保存在数据库，使用任务调度系统扫描然后发送也是可以实现的。

服务端流控（Flow Control)

RabbitMQ 会在启动时检测机器的物理内存数值。默认当 MQ 占用 40% 以上内存时，MQ 会主动抛出一个内存警告并阻塞所有连接（Connections）。可以通过修改 rabbitmq.config 文件来调整内存阈值，默认值是 0.4，如下所示：

[{rabbit, [{vm_memory_high_watermark, 0.4}]}].

默认情况，如果剩余磁盘空间在 1GB 以下，RabbitMQ 主动阻塞所有的生产者。这个阈值也是可调的。

注意：调整队列长度只在消息堆积的情况下有意义，而且会删除先入队的消息，不能实现服务端限流。

消费端限流

在AutoACK为false的情况下，如果一定数目的消息（通过基于consumer或者channel设置Qos的值）未被确认前，不进行消费新的消息。类似ActiveMQ的 FetchSize预取大小，这里也有这么一个概念。

channel.basicQos(2); // 如果超过2条消息没有发送ACK，当前消费者不再接受队列消息
channel.basicConsume(QUEUE_NAME, false, consumer);

Spring 集成 RabbitMQ

引入pom

<!--rabbitmq依赖 --><dependency><groupId>org.springframework.amqp</groupId><artifactId>spring-rabbit</artifactId><version>1.3.5.RELEASE</version></dependency>

引入applicationContext.xml

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:context="http://www.springframework.org/schema/context"xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-3.1.xsdhttp://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context-3.1.xsd"><import resource="classpath*:rabbitMQ.xml" /><!-- 扫描指定package下所有带有如 @Controller,@Service,@Resource 并把所注释的注册为Spring Beans --><context:component-scan base-package="com.example.*" /><!-- 激活annotation功能 --><context:annotation-config /><!-- 激活annotation功能 --><context:spring-configured />
</beans>

引入rabbitMQ.xml

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:rabbit="http://www.springframework.org/schema/rabbit"xsi:schemaLocation="http://www.springframework.org/schema/beanshttp://www.springframework.org/schema/beans/spring-beans-3.0.xsdhttp://www.springframework.org/schema/rabbithttp://www.springframework.org/schema/rabbit/spring-rabbit-1.2.xsd"><!--配置connection-factory，指定连接rabbit server参数 --><rabbit:connection-factory id="connectionFactory" class="org.springframework.amqp.rabbit.connection.CachingConnectionFactory" virtual-host="/" username="guest" password="guest" host="127.0.0.1" port="5672" /><!--通过指定下面的admin信息，当前producer中的exchange和queue会在rabbitmq服务器上自动生成 --><rabbit:admin id="connectAdmin" connection-factory="connectionFactory" /><!--######分隔线######--><!--定义queue --><rabbit:queue name="MY_FIRST_QUEUE" durable="true" auto-delete="false" exclusive="false" declared-by="connectAdmin" /><!--定义direct exchange，绑定MY_FIRST_QUEUE --><rabbit:direct-exchange name="MY_DIRECT_EXCHANGE" durable="true" auto-delete="false" declared-by="connectAdmin"><rabbit:bindings><rabbit:binding queue="MY_FIRST_QUEUE" key="FirstKey"></rabbit:binding></rabbit:bindings></rabbit:direct-exchange><!--定义rabbit template用于数据的接收和发送 --><rabbit:template id="amqpTemplate" connection-factory="connectionFactory" exchange="MY_DIRECT_EXCHANGE" /><!--消息接收者 --><bean id="messageReceiver" class="com.example.rabbitmq.consumer.FirstConsumer"></bean><!--queue listener 观察 监听模式 当有消息到达时会通知监听在对应的队列上的监听对象 --><rabbit:listener-container connection-factory="connectionFactory"><rabbit:listener queues="MY_FIRST_QUEUE" ref="messageReceiver" /></rabbit:listener-container><!--定义queue --><rabbit:queue name="MY_SECOND_QUEUE" durable="true" auto-delete="false" exclusive="false" declared-by="connectAdmin" /><!-- 将已经定义的Exchange绑定到MY_SECOND_QUEUE，注意关键词是key --><rabbit:direct-exchange name="MY_DIRECT_EXCHANGE" durable="true" auto-delete="false" declared-by="connectAdmin"><rabbit:bindings><rabbit:binding queue="MY_SECOND_QUEUE" key="SecondKey"></rabbit:binding></rabbit:bindings></rabbit:direct-exchange><!-- 消息接收者 --><bean id="receiverSecond" class="com.example.rabbitmq.consumer.SecondConsumer"></bean><!-- queue litener 观察 监听模式 当有消息到达时会通知监听在对应的队列上的监听对象 --><rabbit:listener-container connection-factory="connectionFactory"><rabbit:listener queues="MY_SECOND_QUEUE" ref="receiverSecond" /></rabbit:listener-container><!--######分隔线######--><!--定义queue --><rabbit:queue name="MY_THIRD_QUEUE" durable="true" auto-delete="false" exclusive="false" declared-by="connectAdmin" /><!-- 定义topic exchange，绑定MY_THIRD_QUEUE，注意关键词是pattern --><rabbit:topic-exchange name="MY_TOPIC_EXCHANGE" durable="true" auto-delete="false" declared-by="connectAdmin"><rabbit:bindings><rabbit:binding queue="MY_THIRD_QUEUE" pattern="#.Third.#"></rabbit:binding></rabbit:bindings></rabbit:topic-exchange><!--定义rabbit template用于数据的接收和发送 --><rabbit:template id="amqpTemplate2" connection-factory="connectionFactory" exchange="MY_TOPIC_EXCHANGE" /><!-- 消息接收者 --><bean id="receiverThird" class="com.example.rabbitmq.consumer.ThirdConsumer"></bean><!-- queue litener 观察 监听模式 当有消息到达时会通知监听在对应的队列上的监听对象 --><rabbit:listener-container connection-factory="connectionFactory"><rabbit:listener queues="MY_THIRD_QUEUE" ref="receiverThird" /></rabbit:listener-container><!--######分隔线######--><!--定义queue --><rabbit:queue name="MY_FOURTH_QUEUE" durable="true" auto-delete="false" exclusive="false" declared-by="connectAdmin" /><!-- 定义fanout exchange，绑定MY_FIRST_QUEUE 和 MY_FOURTH_QUEUE --><rabbit:fanout-exchange name="MY_FANOUT_EXCHANGE" auto-delete="false" durable="true" declared-by="connectAdmin" ><rabbit:bindings><rabbit:binding queue="MY_FIRST_QUEUE"></rabbit:binding><rabbit:binding queue="MY_FOURTH_QUEUE"></rabbit:binding></rabbit:bindings></rabbit:fanout-exchange><!-- 消息接收者 --><bean id="receiverFourth" class="com.example.rabbitmq.consumer.FourthConsumer"></bean><!-- queue litener 观察 监听模式 当有消息到达时会通知监听在对应的队列上的监听对象 --><rabbit:listener-container connection-factory="connectionFactory"><rabbit:listener queues="MY_FOURTH_QUEUE" ref="receiverFourth" /></rabbit:listener-container>
</beans>

定义生产者

package com.example.rabbitmq.producer;import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.stereotype.Service;@Service
public class MessageProducer {private Logger logger = LoggerFactory.getLogger(MessageProducer.class);@Autowired@Qualifier("amqpTemplate")private AmqpTemplate amqpTemplate;@Autowired@Qualifier("amqpTemplate2")private AmqpTemplate amqpTemplate2;/*** 演示三种交换机的使用** @param message*/public void sendMessage(Object message) {logger.info("Send message:" + message);// amqpTemplate 默认交换机 MY_DIRECT_EXCHANGE// amqpTemplate2 默认交换机 MY_TOPIC_EXCHANGE// Exchange 为 direct 模式，直接指定routingKeyamqpTemplate.convertAndSend("FirstKey", "[Direct,FirstKey] "+message);amqpTemplate.convertAndSend("SecondKey", "[Direct,SecondKey] "+message);// Exchange模式为topic，通过topic匹配关心该主题的队列amqpTemplate2.convertAndSend("msg.Third.send","[Topic,msg.Third.send] "+message);// 广播消息，与Exchange绑定的所有队列都会收到消息，routingKey为空amqpTemplate2.convertAndSend("MY_FANOUT_EXCHANGE",null,"[Fanout] "+message);}
}

定义消费者

package com.example.rabbitmq.consumer;import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageListener;public class FirstConsumer implements MessageListener {private Logger logger = LoggerFactory.getLogger(FirstConsumer.class);public void onMessage(Message message) {logger.info("The first consumer received message : " + message.getBody());}
}

ConnectionFactory：

Spring AMQP 的连接工厂接口，用于创建连接。CachingConnectionFactory 是ConnectionFactory 的一个实现类。

RabbitAdmin：

RabbitAdmin 是 AmqpAdmin 的实现，封装了对 RabbitMQ 的基础管理操作，比如对交换机、队列、绑定的声明和删除等。为什么我们在配置文件（Spring）或者配置类（SpringBoot）里面定义了交换机、队列、绑定关系，并没有直接调用 Channel 的 declare 的方法，Spring 在启动的时候就可以帮我们创建这些元数据？这些事情就是由 RabbitAdmin 完成的。RabbitAdmin 实 现 了 InitializingBean 接 口 ， 里 面 有 唯 一 的 一 个 方 法afterPropertiesSet()，这个方法会在 RabbitAdmin 的属性值设置完的时候被调用。在 afterPropertiesSet ()方法中，调用了一个 initialize()方法。这里面创建了三个Collection，用来盛放交换机、队列、绑定关系。最后依次声明返回类型为 Exchange、Queue 和 Binding 这些 Bean，底层还是调用了 Channel 的 declare 的方法。

Message：

Message 是 Spring AMQP 对消息的封装。两个重要的属性：

body：消息内容。
messageProperties：消息属性。
RabbitTemplate 消息模板：

RabbitTemplate 是 AmqpTemplate 的一个实现（目前为止也是唯一的实现），用来简化消息的收发，支持消息的确认（Confirm）与返回（Return）。跟 JDBCTemplate一 样 ， 它 封 装 了 创 建 连 接 、 创 建 消 息 信 道 、 收 发 消 息 、 消 息 格 式 转 换（ConvertAndSend→Message）、关闭信道、关闭连接等等操作。针对于多个服务器连接，可以定义多个 Template。可以注入到任何需要收发消息的地方使用。

Messager Listener 消息 侦听：

MessageListener 是 Spring AMQP 异步消息投递的监听器接口，它只有一个方法onMessage，用于处理消息队列推送来的消息，作用类似于 Java API 中的 Consumer。

MessageListenerContainer：

MessageListenerContainer可以理解为MessageListener的容器，一个Container只有一个 Listener，但是可以生成多个线程使用相同的 MessageListener 同时消费消息。Container 可以管理 Listener 的生命周期，可以用于对于消费者进行配置。例如：动态添加移除队列、对消费者进行设置，例如 ConsumerTag、Arguments、并发、消费者数量、消息确认模式等等。

转换器 MessageConvertor：

MessageConvertor 的 作用？RabbitMQ 的消息在网络传输中需要转换成 byte[]（字节数组）进行发送，消费者需要对字节数组进行解析。在 Spring AMQP 中，消息会被封装为 org.springframework.amqp.core.Message对象。消息的序列化和反序列化，就是处理 Message 的消息体 body 对象。如果消息已经是 byte[]格式，就不需要转换。如果是 String，会转换成 byte[]。如果是 Java 对象，会使用 JDK 序列化将对象转换为 byte[]（体积大，效率差）。在 调 用 RabbitTemplate 的 convertAndSend() 方 法 发 送 消 息 时 ， 会 使 用MessageConvertor 进行消息的序列化，默认使用 SimpleMessageConverter。在某些情况下，我们需要选择其他的高效的序列化工具。如果我们不想在每次发送消息时自己处理消息，就可以直接定义一个 MessageConvertor。如何 自定义 MessageConverter ？例如：我们要使用 Gson 格式化消息：创建一个类，实现 MessageConverter 接口，重写 toMessage()和 fromMessage()方法。

SpringBoot 集成 RabbitMQ