消息中间件

RabbitMQ

使用

1. 导入依赖

2. 生产者

   • 新建Connection工厂,设置参数
   • 新建Connection连接
   • 新建channel
   • 声明一个队列
   • 使用basicPublish发布消息

package com.oddfar.one;

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

/**
 * @author zhiyuan
 */
public class Producer {
    private final static String QUEUE_NAME = "hello";

    public static void main(String[] args) throws Exception {
        //创建一个连接工厂
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("42.192.149.71");
        factory.setUsername("admin");
        factory.setPassword("123456");
        //channel 实现了自动 close 接口 自动关闭 不需要显示关闭
        //创建连接
        Connection connection = factory.newConnection();
        //获取信道
        Channel channel = connection.createChannel();
        /**
         * 生成一个队列
         * 1.队列名称
         * 2.队列里面的消息是否持久化 也就是是否用完就删除
         * 3.该队列是否只供一个消费者进行消费 是否进行共享 true 可以多个消费者消费
         * 4.是否自动删除 最后一个消费者端开连接以后 该队列是否自动删除 true 自动删除
         * 5.其他参数
         */
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        String message = "hello world";
        /**
         * 发送一个消息
         * 1.发送到那个交换机
         * 2.路由的 key 是哪个
         * 3.其他的参数信息
         * 4.发送消息的消息体
         */
        channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
        System.out.println("消息发送完毕");

    }
    
}

3. 消费者
   • 新建Connection工厂,设置参数
   • 新建Connection连接
   • 新建channel
   • 使用basicConsume发布消息

package com.oddfar.one;

import com.rabbitmq.client.*;

/**
 * @author zhiyuan
 */
public class Consumer {
    private final static String QUEUE_NAME = "hello";

    public static void main(String[] args) throws Exception {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("42.192.149.71");
        factory.setUsername("admin");
        factory.setPassword("123456");
        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();

        System.out.println("等待接收消息.........");

        //推送的消息如何进行消费的接口回调
        DeliverCallback deliverCallback = (consumerTag, delivery) -> {
            String message = new String(delivery.getBody());
            System.out.println(message);
        };
        //取消消费的一个回调接口 如在消费的时候队列被删除掉了
        CancelCallback cancelCallback = (consumerTag) -> {
            System.out.println("消息消费被中断");
        };
        /**
         * 消费者消费消息 - 接受消息
         * 1.消费哪个队列
         * 2.消费成功之后是否要自动应答 true 代表自动应答 false 手动应答
         * 3.消费者未成功消费的回调
         * 4.消息被取消时的回调
         */
        channel.basicConsume(QUEUE_NAME, true, deliverCallback, cancelCallback);
    }

}

消息应答(消费者)

1.自动应答,队列发出消息后默认收到,删除消息

2.手动应答,需要消费者在接收消息回调中使用basicACK确认消息,确认后才删除消息

• Channel.basicAck(用于肯定确认)

  RabbitMQ 已知道该消息并且成功的处理消息，可以将其丢弃了

• Channel.basicNack(用于否定确认)

• Channel.basicReject(用于否定确认)

  与 Channel.basicNack 相比少一个参数，不处理该消息了直接拒绝，可以将其丢弃了

持久化

交换机持久化,声明时设置

队列持久化,声明时设置

//让队列持久化
boolean durable = true;
//声明队列
channel.queueDeclare(TASK_QUEUE_NAME, durable, false, false, null);

消息持久化,发送消息设置参数

不公平分发

给channel设置参数

//不公平分发
//能处理的消息个数
int prefetchCount = 1;
channel.basicQos(prefetchCount);
//采用手动应答
boolean autoAck = false;
channel.basicConsume(TASK_QUEUE_NAME, autoAck, deliverCallback, cancelCallback);

prefetchCount预取值

仍未ack的个数,到达上限先不给它分配

交换机

创建交换机

Channel channel = RabbitMqUtils.getChannel();
channel.exchangeDeclare(EXCHANGE_NAME, "fanout");


String queueName = channel.queueDeclare().getQueue();
//把该临时队列绑定我们的 exchange 其中 routingkey(也称之为 binding key)为空字符串
//第三个参数为routingkey
 channel.queueBind(queueName, EXCHANGE_NAME, "");

类型

1.fanout发送到他绑定的所有队列

2.direct发送到他绑定的指定队列,根据生产者指定的bindingKey路由

channel.basicPublish(EXCHANGE_NAME, bindingKey, null, message.getBytes("UTF-8"));

3.topic主要是设置的bindingKey可以为一个单词列表，以点号分隔开,根据生产者指定的bindingKey路由

其中bindingKey

• ***(星号)可以代替一个单词**
• #(井号)可以替代零个或多个单词

死信队列

死信的来源

• 消息 TTL 过期

  TTL是Time To Live的缩写, 也就是生存时间

  生产者可以设置TTL属性

  Channel channel = RabbitMqUtils.getChannel();
    
  channel.exchangeDeclare(NORMAL_EXCHANGE, BuiltinExchangeType.DIRECT);
  //设置消息的 TTL 时间 10s
  AMQP.BasicProperties properties = new AMQP.BasicProperties().builder().expiration("10000").build();
  //该信息是用作演示队列个数限制
  for (int i = 1; i < 11; i++) {
      String message = "info" + i;
      channel.basicPublish(NORMAL_EXCHANGE, "zhangsan", properties, message.getBytes());
      System.out.println("生产者发送消息:" + message);
  }

• 队列达到最大长度

  队列满了，无法再添加数据到 mq 中

  在创建队列是声明最大长度

  

• 消息被拒绝

  (basic.reject 或 basic.nack) 并且 requeue=false.

channel.basicReject(delivery.getEnvelope().getDeliveryTag(), false);

创建

和普通交换机和队列一样

多了队列出现死信时应该发到哪个交换机

//声明死信和普通交换机 类型为 direct
       channel.exchangeDeclare(NORMAL_EXCHANGE, BuiltinExchangeType.DIRECT);
       channel.exchangeDeclare(DEAD_EXCHANGE, BuiltinExchangeType.DIRECT);

       //声明死信队列
       String deadQueue = "dead-queue";
       channel.queueDeclare(deadQueue, false, false, false, null);
       //死信队列绑定：队列、交换机、路由键（routingKey）
       channel.queueBind(deadQueue, DEAD_EXCHANGE, "lisi");

       //正常队列绑定死信队列信息
       Map<String, Object> params = new HashMap<>();
       //正常队列设置死信交换机 参数 key 是固定值
       params.put("x-dead-letter-exchange", DEAD_EXCHANGE);
       //正常队列设置死信 routing-key 参数 key 是固定值
       params.put("x-dead-letter-routing-key", "lisi");

       //正常队列
       String normalQueue = "normal-queue";
       channel.queueDeclare(normalQueue, false, false, false, params);
       channel.queueBind(normalQueue, NORMAL_EXCHANGE, "zhangsan");

延迟队列

• 队列设置TTL

  过期就扔到死信队列

  Map<String, Object> args = new HashMap<>(3);
         //声明当前队列绑定的死信交换机
         args.put("x-dead-letter-exchange", Y_DEAD_LETTER_EXCHANGE);
         //声明当前队列的死信路由 key
         args.put("x-dead-letter-routing-key", "YD");
         //声明队列的 TTL
         args.put("x-message-ttl", 10000);
         return QueueBuilder.durable(QUEUE_A).withArguments(args).build();

  

• 消息设置TTL

  即将投递到消费者之前判定

同时设置使用较小值

插件

可以设置一个延迟交换机