rabbitmq入门-Hello World

rabbitMQ是一个在AMQP基础上完整的，可复用的企业消息系统。rabbitMQ使用二郎语言（ERLANG）编写，使用需安装erlang。

 

rabbitmq的三个名词解释

生产者：发送消息的程序就是一个生产者(producer)

队列：队列是发送消息的缓冲，基本上是一个无限的缓冲。多个生产者（producers）能够把消息发送给同一个队列，同样，多个消费者（consumers）也能攻从一个队列（queue）中获取数据。

消费：获取队列中的消息。

 

Hello World步骤

客户端

第一步，建立一个到RabbitMQ服务器的连接

 

ConnectionFactory factory = new ConnectionFactory();
factory.setHost("localhost");
Connection connection = factory.newConnection();

 

 第二步：定义queue名称

 

 

Channel channel = connection.createChannel();
channel.queueDeclare(QUEUE_NAME, false, false, false, null);

 定义QUEUE_NAME，消息将通过定义的queue发送消息

 

第三部：发送消息

String message = "Hello World!";
channel.basicPublish("", QUEUE_NAME, null, message.getBytes());

 发送完消息，切记关闭连接

channel.close();
connection.close();

 

服务端

 

第一、二步：跟客户端一致，定义域rabbitmq服务器的连接并指定读取的queue名称

ConnectionFactory factory = new ConnectionFactory();
factory.setHost("localhost");
Connection connection = factory.newConnection();
Channel channel = connection.createChannel();
		
channel.queueDeclare(QUEUE_NAME, false, false, false, null);

 第三步：定义消费者进行消费

QueueingConsumer consumer = new QueueingConsumer(channel);
channel.basicConsume(QUEUE_NAME, true, consumer);

while (true) {
  QueueingConsumer.Delivery delivery = consumer.nextDelivery();
  String message = new String(delivery.getBody());
  System.out.println(" [x] Received '" + message + "'");
}

 消费者根据queue名称读取消息

 

整合代码

客户端：

public class MQClient 
{
	private final static String QUEUE_NAME="hello";
    public static void main( String[] args ) throws IOException
    {
       ConnectionFactory factory = new ConnectionFactory();
       factory.setHost("localhost");
       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(" [x] Sent '" + message + "'");
       
       channel.close();
       connection.close();
    }
}

输出：

 [x] Sent 'Hello World!'

  

服务端

public class MQServer {
	private final static String QUEUE_NAME = "hello";
	public static void main(String[] args) throws IOException, ShutdownSignalException, ConsumerCancelledException, InterruptedException {
		ConnectionFactory factory = new ConnectionFactory();
		factory.setHost("localhost");
		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) {
	      QueueingConsumer.Delivery delivery = consumer.nextDelivery();
	      String message = new String(delivery.getBody());
	      System.out.println(" [x] Received '" + message + "'");
	    }
		
	}
}

 输出：

 [*] Waiting for messages. To exit press CTRL+C
 [x] Received 'Hello World!'