RabbitMq helloWorld

发布日期: 2021-10-20

更新日期: 2021-10-27

文章字数: 1.5k

阅读时长: 6 分

阅读次数: loading

HelloWorld

一个mq服务可以设置多个虚拟机，每个虚拟机就像阿当予一个独立的mq.

build by maven

pom.xml

<dependency>
  <groupId>com.rabbitmq</groupId>
  <artifactId>amqp-client</artifactId>
  <version>${last-version}/version>
</dependency>

producer

//通过连接工厂穿件新的连接和mq建立连接
      ConnectionFactory connectionFactory = new ConnectionFactory();
      connectionFactory.setHost("127.0.0.1");
      connectionFactory.setPort(5672);
      connectionFactory.setUsername("guest");
      connectionFactory.setPassword("guest");
      //设置虚拟机,一个mq的服务可以设置多个虚拟机，每一个虚拟机就相当于一个独立的mq
      connectionFactory.setVirtualHost("/");
      //建立新连接
      Connection connection = null;
      Channel channel = null;
      try {
          connection = connectionFactory.newConnection();
          //设置会话通道,生产者和mq服务所有通信都在channel通道中完成。
          channel = connection.createChannel();
          //声明队列，如果队列在mq中没有则要创建
          //String queue, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments
          /**
           * 1.queue 队列名称
           * 2.durable 是否持久化，如果持久化，mq重启队列还在
           * 3.exclusive 是否独占连接， 队列只允许在该连接中访问，如果connection连接关闭队列自动删除， 如果将此参数设置true可用于临时队列的创建。
           * 4.autoDelete 自动删除，对了不再使用时是否自动删除此队列，如果将此参数和exclusive参数设置为true就可以实现临时队列（队列不用了就自动删除）
           * 5.argument是参数，可以设置一个队列的扩展参数，（如存活时间）
           */
          channel.queueDeclare(QUEUE, true, false, false ,null);
          //发送消息
          //String exchange, String routingKey, BasicProperties props, byte[] body)
          /**
           * 1. exchange, 交换机，如果不指定将使用mq默认的交换机（设置为“”）
           * 2. routingKey， 路由key，交换机Kenji路由key来讲消息转发到指定的队列，如果使用默认交换机，routingKey这是为队列的名称。
           * 3. props 消息属性
           * 4.body 消息内容
           */
          String message = "helloWorld !!!!";
          channel.basicPublish("", QUEUE, null, message.getBytes());
      } catch (IOException e) {
          e.printStackTrace();
      } catch (TimeoutException e) {
          e.printStackTrace();
      }finally {
          //关闭连接
          //先关闭通道
          try {
              channel.close();

          } catch (IOException e) {
              e.printStackTrace();
          } catch (TimeoutException e) {
              e.printStackTrace();
          }

          try {
              connection.close();
          } catch (IOException e) {
              e.printStackTrace();
          }

      }

consumer

//通过连接工厂穿件新的连接和mq建立连接
     ConnectionFactory connectionFactory = new ConnectionFactory();
     connectionFactory.setHost("127.0.0.1");
     connectionFactory.setPort(5672);
     connectionFactory.setUsername("guest");
     connectionFactory.setPassword("guest");
     //设置虚拟机,一个mq的服务可以设置多个虚拟机，每一个虚拟机就相当于一个独立的mq
     connectionFactory.setVirtualHost("/");
     //建立新连接
     try {
         Connection connection = connectionFactory.newConnection();
         //设置会话通道,生产者和mq服务所有通信都在channel通道中完成。
         Channel channel = connection.createChannel();

         //声明队列，如果队列在mq中没有则要创建
         //String queue, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments
         /**
          * 1.queue 队列名称
          * 2.durable 是否持久化，如果持久化，mq重启队列还在
          * 3.exclusive 是否独占连接， 队列只允许在该连接中访问，如果connection连接关闭队列自动删除， 如果将此参数设置true可用于临时队列的创建。
          * 4.autoDelete 自动删除，对了不再使用时是否自动删除此队列，如果将此参数和exclusive参数设置为true就可以实现临时队列（队列不用了就自动删除）
          * 5.argument是参数，可以设置一个队列的扩展参数，（如存活时间）
          */
         channel.queueDeclare(QUEUE, true, false, false ,null);
         DefaultConsumer defaultConsumer = new DefaultConsumer(channel){
             /**
              * 当接收到消息时，此方法被调用。
              * @param consumerTag 消费者标签，用来标识消费者， 在监听队列式设置channel。basicConsumer
              * @param envelope 信封，通过envelope
              * @param properties
              * @param <body></body>
              * @throws IOException
              */
             @Override
             public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                 //交换机
                 String exchange = envelope.getExchange();
                 //消息id，mq在channel用来标识消费的id，可用于确认消息已接收。
                 long deliveryTag = envelope.getDeliveryTag();
                 //消息的属性
                 String message = new String(body, "utf-8");
                 System.out.println("receive message:"+ message);

             }
         };

         //监听队列
         //String queue, boolean autoAck, Consumer callback
         /**
          * 1. queue 队列名称
          * 2.autoAck 自动回复，当消费者接收到消息后要告诉mq消息已收到，如果将此参数设置为true表示会自动回复mq，如果设置为false要通过编程实现回复。
          * 3.callback，消费方法，当消费者接收到消息要执行的方法。
          */
        channel.basicConsume(QUEUE, true, defaultConsumer);

     } catch (IOException e) {
         e.printStackTrace();
     } catch (TimeoutException e) {
         e.printStackTrace();
     }

工作模式

hello world

一个生产者 > 个队列 > 一个消费者

work queues 工作队列模式

和helloworld 的区别是一个队列有多个消费者，但是消息不会被重复消费，轮询的方式将消息平均发送给消费者。

发布订阅模式

生产者 > 交换机 > 各个队列 > 每个队列有不同消费者
一个消息一下子被多个消费者接收到。

不同点：

type = FANOUT
交换机将消息发送到所有绑定了的队列中，所有队列默认一个相同的 routingKey

路由模式

和发布订阅模式相比，交换机可以根据路由key，将消息发送到指定的队列

不同点：

type = direct
queue 绑定自定义 routingKey

通配符模式

不同点：

type = topic
routingKey 通配符匹配即可配对成功

“#”:匹配一个或多个词，比如inform。# 可匹配 inform.sms、inform.email、inform.emal.sms
““: 只能匹配一个词，比如inform., 可以匹配inform.sms、 inform.email

不同点：