RabbitMQ3.7快速上手指南：深入研究与实践

按照官方教程(http://www.rabbitmq.com/getstarted.html)测试hello world:

搭建环境

1&＃xff09;java client

生产者和消费者都属于客户端&＃xff0c;rabbitMQ的java客户端如下&＃xff1a;

我们先用 rabbitMQ官方提供的java client测试&＃xff0c;目的是对RabbitMQ的交互过程有个清晰的认识。

参考 &＃xff1a;https://github.com/rabbitmq/rabbitmq-java-client/

2&＃xff09;创建maven工程

创建生产者工程和消费者工程&＃xff0c;分别加入RabbitMQ java client的依赖。

test-rabbitmq-producer&＃xff1a;生产者工程

test-rabbitmq-consumer&＃xff1a;消费者工程

com.rabbitmqamqp-client4.0.3

org.springframework.bootspring-boot-starter-logging



生产者

生产者操作流程如下&＃xff1a;

1&＃xff09;创建连接

2&＃xff09;创建通道

3&＃xff09;声明队列

4&＃xff09;发送消息

在生产者工程下的test包中创建测试类如下&＃xff1a;

public class Producer01 {//队列名称private static final String QUEUE &＃61; "helloworld";public static void main(String[] args) throws IOException, TimeoutException {Connection connection &＃61; null;Channel channel &＃61; null;try{ConnectionFactory factory &＃61; new ConnectionFactory();factory.setHost("localhost");factory.setPort(5672);factory.setUsername("guest");factory.setPassword("guest");factory.setVirtualHost("/");//rabbitmq默认虚拟机名称为“/”&＃xff0c;虚拟机相当于一个独立的mq服务器//创建与RabbitMQ服务的TCP连接connection &＃61; factory.newConnection();//创建与Exchange的通道&＃xff0c;每个连接可以创建多个通道&＃xff0c;每个通道代表一个会话任务channel &＃61; connection.createChannel();/*** 声明队列&＃xff0c;如果Rabbit中没有此队列将自动创建* param1:队列名称* param2:是否持久化* param3:队列是否独占此连接* param4:队列不再使用时是否自动删除此队列* param5:队列参数*/channel.queueDeclare(QUEUE, true, false, false, null);String message &＃61; "helloworld小明"&＃43;System.currentTimeMillis();/*** 消息发布方法* param1&＃xff1a;Exchange的名称&＃xff0c;如果没有指定&＃xff0c;则使用Default Exchange* param2:routingKey,消息的路由Key&＃xff0c;是用于Exchange&＃xff08;交换机&＃xff09;将消息转发到指定的消息队列* param3:消息包含的属性* param4&＃xff1a;消息体*//*** 这里没有指定交换机&＃xff0c;消息将发送给默认交换机&＃xff0c;每个队列也会绑定那个默认的交换机&＃xff0c;但是不能显示绑定或解除绑定*　默认的交换机&＃xff0c;routingKey等于队列名称*/channel.basicPublish("", QUEUE, null, message.getBytes());System.out.println("Send Message is:&＃39;" &＃43; message &＃43; "&＃39;");}catch(Exception ex){ex.printStackTrace();}finally{if(channel !&＃61; null){channel.close();}if(connection !&＃61; null){connection.close();}}}
}


消费者

消费者操作流程如下&＃xff1a;

1&＃xff09;创建连接

2&＃xff09;创建通道

3&＃xff09;声明队列

4&＃xff09;监听队列

5&＃xff09;接收消息

6&＃xff09;ack回复

在消费者工程下的test包中创建测试类如下&＃xff1a;

public class Consumer01 {private static final String QUEUE &＃61; "helloworld";public static void main(String[] args) throws IOException, TimeoutException {ConnectionFactory factory &＃61; new ConnectionFactory();//设置MabbitMQ所在服务器的ip和端口factory.setHost("127.0.0.1");factory.setPort(5672);Connection connection &＃61; factory.newConnection();Channel channel &＃61; connection.createChannel();//声明队列channel.queueDeclare(QUEUE, true, false, false, null);//定义消费方法DefaultConsumer consumer &＃61; new DefaultConsumer(channel) {/*** 消费者接收消息调用此方法* &＃64;param consumerTag 消费者的标签&＃xff0c;在channel.basicConsume()去指定* &＃64;param envelope 消息包的内容&＃xff0c;可从中获取消息id&＃xff0c;消息routingkey&＃xff0c;交换机&＃xff0c;消息和重传标志(收到消息失败后是否需要重新发送)* &＃64;param properties* &＃64;param body* &＃64;throws IOException*/&＃64;Overridepublic void handleDelivery(String consumerTag,Envelope envelope,AMQP.BasicProperties properties,byte[] body)throws IOException {//交换机String exchange &＃61; envelope.getExchange();//路由keyString routingKey &＃61; envelope.getRoutingKey();//消息idlong deliveryTag &＃61; envelope.getDeliveryTag();//消息内容String msg &＃61; new String(body,"utf-8");System.out.println("receive message.." &＃43; msg);}};/*** 监听队列String queue, boolean autoAck,Consumer callback* 参数明细* 1、队列名称* 2、是否自动回复&＃xff0c;设置为true为表示消息接收到自动向mq回复接收到了&＃xff0c;mq接收到回复会删除消息&＃xff0c;设置为false则需要手动回复* 3、消费消息的方法&＃xff0c;消费者接收到消息后调用此方法*/channel.basicConsume(QUEUE, true, consumer);}
}