欢迎您访问程序员文章站本站旨在为大家提供分享程序员计算机编程知识!
您现在的位置是: 首页

[rabbitmq] 安装与配置 javaexchangeerlang

程序员文章站 2022-05-30 16:17:33
...
RabbitMQ
-------------------------
1
下载

erlang
http://www.erlang.org/download/otp_win32_17.3.exe
http://www.erlang.org/download/otp_src_17.3.tar.gz
http://www.erlang.org/download/otp_doc_html_17.3.tar.gz
http://www.erlang.org/download/otp_doc_man_17.3.tar.gz

rabbitmq server
http://www.rabbitmq.com/releases/rabbitmq-server/v3.4.2/rabbitmq-server-3.4.2.exe

rabbitmq java client
http://www.rabbitmq.com/releases/rabbitmq-java-client/v3.4.2/rabbitmq-java-client-bin-3.4.2.zip
http://www.rabbitmq.com/releases/rabbitmq-java-client/v3.4.2/rabbitmq-java-client-3.4.2.zip

2
配置

3
基本概念
http://www.rabbitmq.com/tutorials/tutorial-one-java.html

RabbitMQ 是一个消息代理,它接收生产者的消息,然后转发给消费者.同时,它路由,缓存,持久这个消息根据设定的规则.
producer 生产者,只负责发送消息.消息发送到指定队列中.
queue 队列或邮箱,它存储消息,它的大小没有限制,只要磁盘空间足够.消息只能缓存在队列中,队列属于消息代理的一部分.消息代理还包括 exchange 等.
consumer 消费者,它从队列接收消息.

AMQP:高级消息队列协议.

P(producer)   -- msg --> |Queue|Queue|Queue|  -- msg --> C(consumer)

PS:
生产者,消息代理,消费者通常不在同一机器上.

4 Hello World
----------------
我们这里使用 Java 客户端.

STEP 1:

    P(producer)   -- msg --> |HelloQueue|


public class P {

  private final static String QUEUE_NAME = "HelloQueue";

  public static void main(String[] args)
      throws Exception {
      // 建立连接
      ConnectionFactory factory = new ConnectionFactory();
      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!";
      
      // 发布消息到 HelloQueue
      channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
      
      // 关闭连接
      channel.close();
      connection.close();
  }

}



STEP 2

    |HelloQueue| -- msg --> C(consumer)

public class C {

  private final static String QUEUE_NAME = "HelloQueue";

  public static void main(String[] args)
      throws Exception {
    
    // 建立连接
    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) {
        // block method ? yes.
        // 阻塞直到接收到下一条消息
        QueueingConsumer.Delivery delivery = consumer.nextDelivery(); 
        String message = new String(delivery.getBody());
        System.out.println("Received : " + message);
    }
}
  



5
Work Queue

    P -- MSG -- |Q| -- MSG -- Worker1|Worker2

Work1 , Work2 轮流获取消息处理,
为避免Worker1消息单节点处理失败, 修改ACK返回时机为消息处理成功后.ACK返回,消息代理才会删除已处理的消息,否则是Worker1一接收到就删除.

    boolean autoAck = false;
    channel.basicConsume("hello", autoAck, consumer);

while (true) {
    QueueingConsumer.Delivery delivery = consumer.nextDelivery();
    //...      
    channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
}



为保证消息代理节点失败,导致消息丢失, 设置消息持久化参数:

boolean durable = true;
channel.queueDeclare("hello", durable, false, false, null);

...

 channel.basicPublish( "", TASK_QUEUE_NAME, 
            MessageProperties.PERSISTENT_TEXT_PLAIN,
            message.getBytes());


6
发布/订阅模式
一个消息多个接收者接收.
EXCHANGE: 消息交换,用于路由/过滤消息.

P -- MSG -- |X| -- |Q1| -- MSG -- C1
                -- |Q2| -- MSG -- C2


     // 交换策略扇出(fanout):消息交换到所有队列 
     channel.exchangeDeclare(EXCHANGE_NAME, "fanout");

     String message = getMessage(argv);

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


     channel.exchangeDeclare(EXCHANGE_NAME, "fanout");
     // 临时队列.随机命名的队列
     String queueName = channel.queueDeclare().getQueue();
     // 绑定交换与队列
     channel.queueBind(queueName, EXCHANGE_NAME, "");



7
路由
绑定: 一个队列对交换的消息感兴趣.
绑定KEY: 绑定时设置
路由KEY: 发布时设置 routingKey, fanout 会简单忽略这个值.
交换策略:
direct exchange: 直接交换,根据 routingKey 来选择发布到哪些队列.比fanout更加灵活.
fanout exchange: 扇出交换(全部队列都会接收这个消息)
topic exchange: routingKey用 "." 号分开.根据 routingKey 来选择发布到哪些匹配的队列, 支持模糊匹配.

     channel.exchangeDeclare(EXCHANGE_NAME, "topic");
     ...
     channel.basicPublish(EXCHANGE_NAME, routingKey, null, message.getBytes());



      channel.exchangeDeclare(EXCHANGE_NAME, "topic");
      String queueName = channel.queueDeclare().getQueue();



      for(String bindingKey : bindingKeys){
          channel.queueBind(queueName, EXCHANGE_NAME, bindingKey);
      }

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

      while (true) {
          QueueingConsumer.Delivery delivery = consumer.nextDelivery();
          String message = new String(delivery.getBody());
          String routingKey = delivery.getEnvelope().getRoutingKey();
          ...
        }


8
RPC 远程过程调用
通常发送方要等待接收方的处理结果.

发送方阻塞直到方法返回结, 每个客户端一个响应队列.
为了识别哪个响应属于哪个请求,要加个请求ID.

result = Client.call(request); // block until result has returned

    Client -- id:RequestId , replyTo:ReplyQueue -- HelloQueue -- Server
           |                                                       |
           -- id:RequestId , replyTo:ReplyQueue -- ReplyQueue ------


RPCServer
private static final String RPC_QUEUE_NAME = "rpc_queue";

ConnectionFactory factory = new ConnectionFactory();
factory.setHost("localhost");

Connection connection = factory.newConnection();
Channel channel = connection.createChannel();

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

channel.basicQos(1);

QueueingConsumer consumer = new QueueingConsumer(channel);
channel.basicConsume(RPC_QUEUE_NAME, false, consumer);

System.out.println(" [x] Awaiting RPC requests");

while (true) {
    QueueingConsumer.Delivery delivery = consumer.nextDelivery();

    BasicProperties props = delivery.getProperties();
    BasicProperties replyProps = new BasicProperties
                                     .Builder()
                                     .correlationId(props.getCorrelationId())
                                     .build();

    String message = new String(delivery.getBody());
    int n = Integer.parseInt(message);

    System.out.println(" [.] fib(" + message + ")");
    String response = "" + fib(n);

    channel.basicPublish( "", props.getReplyTo(), replyProps, response.getBytes());

    channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
}



RPCClient
private Connection connection;
private Channel channel;
private String requestQueueName = "rpc_queue";
private String replyQueueName;
private QueueingConsumer consumer;

public RPCClient() throws Exception {
    ConnectionFactory factory = new ConnectionFactory();
    factory.setHost("localhost");
    connection = factory.newConnection();
    channel = connection.createChannel();

    replyQueueName = channel.queueDeclare().getQueue(); 
    consumer = new QueueingConsumer(channel);
    channel.basicConsume(replyQueueName, true, consumer);
}

public String call(String message) throws Exception {     
    String response = null;
    String corrId = java.util.UUID.randomUUID().toString();

    BasicProperties props = new BasicProperties
                                .Builder()
                                .correlationId(corrId)
                                .replyTo(replyQueueName)
                                .build();

    channel.basicPublish("", requestQueueName, props, message.getBytes());

    while (true) {
        QueueingConsumer.Delivery delivery = consumer.nextDelivery();
        if (delivery.getProperties().getCorrelationId().equals(corrId)) {
            response = new String(delivery.getBody());
            break;
        }
    }

    return response; 
}

public void close() throws Exception {
    connection.close();
}