rabbitmq 简单示例(Hello World)

 

一：消息中间件：

AMQP，即Advanced Message Queuing Protocol，高级消息队列协议，是应用层协议的一个开放标准，为面向消息的中间件设计

 

RabbitMQ是实现AMQP（高级消息队列协议）的消息中间件的一种

 

常见的消息中间件：

Kafka、RabbitMQ、RocketMQ等

 

二：rabbit mq消息中间件：

1.图：

　　

  P：代表生产者，C：代表消费者，中间红色的是消息队列。 

 注：关于消息队列的好处可以去官网了解，可以参考数据结构中队列的特点理解。

 

2.HelloWorld示例的代码：

　　该简单示例的代码在rabbitmq官网也有实现：http://www.rabbitmq.com/tutorials/tutorial-one-java.html

发送者：

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;

public class Send {

//  private final static String QUEUE_NAME = "task_event_queue";
//  private final static String QUEUE_NAME = "task-queue-default";
  private final static String QUEUE_NAME = "task-queue-default_mytest 2018-3-15";

  public static void main(String[] argv) throws Exception {
    ConnectionFactory factory = new ConnectionFactory();
    factory.setHost("127.0.0.1");
    factory.setPort(5672);
    factory.setUsername("guest");
    factory.setPassword("guest");
    
    Connection connection = factory.newConnection();
    Channel channel = connection.createChannel();

    /*
     * queueDeclare方法参数：
     * 	第一个参数表示队列名称、第二个参数为是否持久化（true表示是，服务器重启仍存在此队列）、第三个参数为是否是独占队列（创建者可以使用的私有队列，断开后自动删除）、
     * 	第四个参数为当所有消费者客户端连接断开时是否自动删除队列、第五个参数为队列的其他参数 
     */
    channel.queueDeclare(QUEUE_NAME, true, false, false, null);  // 注：第二个参数要设置为true，之前设置为false，老是报连接超时
    String message = "Hello World!  2018-3-15 14:14:06";
    
    /*
     * basicPublish方法参数：
     * 	第一个参数为交换机名称、第二个参数为队列映射的路由key、第三个参数为消息的其他属性、第四个参数为发送信息的主体
     */
    channel.basicPublish("", QUEUE_NAME, null, message.getBytes("UTF-8")); // 发送消息，使用默认的direct交换器 
    System.out.println(" [x] Sent '" + message + "'");

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

　　

消费者：

import java.io.IOException;

import com.rabbitmq.client.*;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.Consumer;
import com.rabbitmq.client.DefaultConsumer;

public class Receive {
	
	private final static String QUEUE_NAME = "task-queue-default_mytest 2018-3-15";
	
	public static void main(String[] args) throws Exception {
		ConnectionFactory factory = new ConnectionFactory();
	    factory.setHost("127.0.0.1");
	    factory.setUsername("guest");
	    factory.setPassword("guest");
	    Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();
        
//        channel.queueDeclare(QUEUE_NAME, true, false, false, null);  
	    channel.queueDeclare(QUEUE_NAME, true, false, false, null);
	    System.out.println(" [*] Waiting for messages. To exit press CTRL+C");
	    
	    /*
	     * 告诉服务器我们需要那个管道的消息，如果管道中有消息，就会执行回调函数handleDelivery
	     */
	    Consumer consumer = new DefaultConsumer(channel) {
          @Override
          public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body)
              throws IOException {
            String message = new String(body, "UTF-8");
            System.out.println(" [Consumer] Received '" + message + "'");
          }
        };
        
        /*
         * basicConsume方法参数：
         * 	第一个参数是Consumer绑定的队列名
         * 	第二个参数是自动确认标志，如果为true，表示Consumer接受到消息后，会自动发确认消息(Ack消息)给消息队列，消息队列会将这条消息从消息队列里删除
         * 	第三个参数就是Consumer对象，用于处理接收到的消息
         */
        channel.basicConsume(QUEUE_NAME, true, consumer);
        
        channel.close();
        factory.clone();
	}
}