rabbitMQ消息队列入门

消息队列一般是在项目中，将一些无需即时返回且耗时的操作提取出来，进行了异步处理，而这种异步处理的方式大大的节省了服务器的请求响应时间，从而提高了系统的吞吐量。

在项目启动之初来预测将来项目会碰到什么需求，是极其困难的。消息队列在处理过程中间插入了一个隐含的、基于数据的接口层，两边的处理过程都要实现这一接口。这允许你独立的扩展或修改两边的处理过程，只要确保它们遵守同样的接口约束。消息队列可以解决这样一个问题，也就是其解耦性。解耦伴随的好处就是降低冗余，灵活，易于扩展。

峰值处理能力：当你的应用上了Hacker News的首页，你将发现访问流量攀升到一个不同寻常的水平。在访问量剧增的情况下，你的应用仍然需要继续发挥作用，但是这样的突发流量并不常见；如果为以能处理这类峰值访问为标准来投入资源随时待命无疑是巨大的浪费。使用消息队列能够使关键组件顶住增长的访问压力，而不是因为超出负荷的请求而完全崩溃。

新建发送者Send.java，代码如下：

package com.luo.rabbit.test.one;

import com.rabbitmq.client.Channel;

import com.rabbitmq.client.Connection;

import com.rabbitmq.client.ConnectionFactory;

public class Send {

//队列名称

private final static String QUEUE_NAME = "queue";

public static void main(String[] argv) throws java.io.IOException

{

/*** 创建连接连接到MabbitMQ */

ConnectionFactory factory = new ConnectionFactory();

//设置MabbitMQ所在主机ip或者主机名

factory.setHost("127.0.0.1");

//创建一个连接

Connection connection = factory.newConnection();

//创建一个频道

Channel channel = connection.createChannel();

//指定一个队列

channel.queueDeclare(QUEUE_NAME, false, false, false, null);

//发送的消息

String message = "hello world!";

//往队列中发出一条消息

channel.basicPublish("", QUEUE_NAME, null, message.getBytes());

System.out.println("Sent '"+ message + "'");

//关闭频道和连接

channel.close();

connection.close();

}

}

新建接收者Recv.java，代码如下：

package com.luo.rabbit.test.one;

import com.rabbitmq.client.Channel;

import com.rabbitmq.client.Connection;

import com.rabbitmq.client.ConnectionFactory;

import com.rabbitmq.client.QueueingConsumer;

public class Recv {

//队列名称

private final static String QUEUE_NAME = "queue";

public static void main(String[] argv) throws java.io.IOException, java.lang.InterruptedException

{

//打开连接和创建频道，与发送端一样

ConnectionFactory factory = new ConnectionFactory();

//设置MabbitMQ所在主机ip或者主机名

factory.setHost("127.0.0.1");

Connection connection = factory.newConnection();

Channel channel = connection.createChannel();

//声明队列，主要为了防止消息接收者先运行此程序，队列还不存在时创建队列。

channel.queueDeclare(QUEUE_NAME, false, false, false, null);

System.out.println("Waiting for messages. To exit press CTRL+C");

//创建队列消费者

QueueingConsumer consumer = new QueueingConsumer(channel);

//指定消费队列

channel.basicConsume(QUEUE_NAME, true, consumer);

while(true)

{

//nextDelivery是一个阻塞方法（内部实现其实是阻塞队列的take方法）

QueueingConsumer.Delivery delivery = consumer.nextDelivery();

String message = new String(delivery.getBody());

System.out.println("Received '"+ message + "'");

}

}

}