【RabbitMQ】安装、配置、初体验

生产环境是Ubuntu LTS 14.04

首先 rabbitmq是基于 Erlang 语言

apt-get install erlang

然后安装rabbitmq

apt-get install rabbitmq-server

配置rabbitmq维护

rabbitmq-plugins enable rabbitmq_management

启动rabbitmq

重启 service rabbitmq-server restart 停止 service rabbitmq-server stop 启动 service rabbitmq-server start

使用默认配置即可访问 http://yourhost:15672/ 用户名和密码都是 guest 如果不能访问则是环境存在问题

介绍RabbitMQ

RabbitMQ 是信息传输的中间者。本质上，他从生产者（producers）接收消息，转发这些消息给消费者（consumers）.换句话说，他能够按根据你指定的规则进行消息转发、缓冲、和持久化。 RabbitMQ 的一些常见的术语： Producing意味着无非是发送。一个发送消息的程序是一个producer(生产者)。一般用下图表示Producer： Queue（队列）类似邮箱。依存于RabbitMQ内部。虽然消息通过RabbitMQ在你的应用中传递，但是它们只能存储在queue中。队列不受任何限制，可以存储任何数量的消息—本质上是一个无限制的缓存。很多producers可以通过同一个队列发送消息，相同的很多consumers可以从同一个队列上接收消息。一般用下图表示队列： Consuming（消费）类似于接收。consumer是基本属于等待接收消息的程序。一般使用下图表示Consumer:

抽象流程：生产者生产消息进入消息队列被消费者消费掉 注意：producer（生产者）,consumer（消费者）,broker（RabbitMQ服务）并不需要部署在同一台机器上，实际上在大多数实际的应用中，也不会部署在同一台机器上。

java例子

先导入jar包，Maven引入，大家也可以去下载jar包

<!-- https://mvnrepository.com/artifact/com.rabbitmq/amqp-client --> <dependency> <groupId>com.rabbitmq</groupId> <artifactId>amqp-client</artifactId> <version>4.1.0</version> </dependency>

Client端（生产者）

import java.io.IOException; import java.util.concurrent.TimeoutException; import com.rabbitmq.client.Channel; import com.rabbitmq.client.Connection; import com.rabbitmq.client.ConnectionFactory; /** * Client端 * @author xiaoping * */ public class Client { //队列名 private final static String QUEUE_NAME = "FIRST_QUEUE"; public static void main(String[] args) throws IOException, TimeoutException { //1.建立连接工厂对象 ConnectionFactory connectionFactory = new ConnectionFactory(); connectionFactory.setHost("115.159.142.241"); connectionFactory.setUsername("guest"); connectionFactory.setPassword("guest"); //2.建立一个连接 Connection connection = connectionFactory.newConnection(); //3.创建一个频道 Channel channel = connection.createChannel(); //4.制定一个队列 channel.queueDeclare(QUEUE_NAME, false, false, false, null); //5.发送消息 String msg = "hello world!"; channel.basicPublish("", QUEUE_NAME, null,msg.getBytes()); System.out.println("发送消息：" + msg); //6.关闭频道和连接 channel.close(); connection.close(); } }

Rev端（消费者）

import java.util.concurrent.TimeoutException; import com.rabbitmq.client.Channel; import com.rabbitmq.client.Connection; import com.rabbitmq.client.ConnectionFactory; import com.rabbitmq.client.QueueingConsumer; /** * 接收端 * @author xiaoping * */ public class Rev { //队列名称 private final static String QUEUE_NAME = "FIRST_QUEUE"; public static void main(String[] argv) throws java.io.IOException, java.lang.InterruptedException, TimeoutException{ //打开连接和创建频道，与发送端一样 ConnectionFactory factory = new ConnectionFactory(); factory.setHost("115.159.142.241"); factory.setUsername("guest"); factory.setPassword("guest"); Connection connection = factory.newConnection(); Channel channel = connection.createChannel(); //声明队列，主要为了防止消息接收者先运行此程序，队列还不存在时创建队列。 channel.queueDeclare(QUEUE_NAME, false, false, false, null); System.out.println("等待消息Client端发来消息..."); //创建队列消费者 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(" 接收到Client端消息'" + message + "'"); } } }

这里生产者每次发送一个消息，都会进入消息队列，一次被我们的消费者消费掉。