Spring中集成RabbitMQ中的消息队列跟发布订阅
Alan 五月 4th, 2019 1.Java语言 Leave a comment
1.介绍
RabbitMQ是一个消息代理：它接受和转发消息。您可以将其视为邮局：当您将要发布的邮件放在邮箱中时，您可以确定邮件先生或Mailperson女士最终会将邮件发送给您的收件人。在这个类比中，RabbitMQ是一个邮箱，邮局和邮递员。

RabbitMQ和邮局之间的主要区别在于它不处理纸张，而是接受，存储和转发二进制数据 – 消息。

RabbitMQ和一般的消息传递使用了一些术语。

1.生产只意味着发送，发送消息的程序是生产者：

1556961989-3296-producer
2.queue是RabbitMQ中的邮箱的名称，虽然信息流经RabbitMQ和应用程序，但它们只能存储在queue中。queue仅由主机的存储器＆磁盘限制约束，它本质上就是一个大的消息缓冲器。许多生产者可以发送消息到一个队列，并且许多消费者可以尝试从一个队列接收数据。这就是我们代表队列的方式：
1556961989-3357-queue
3.消费消息与接收消息有类似的意义。一个消费者是一个程序，主要是等待接收信息：
1556961990-9390-consumer
请注意，生产者，消费者和代理不必驻留在同一主机上; 实际上在大多数应用中他们没有。应用程序也可以是生产者和消费者。

2.工作队列（Queue）
1556963391-5142-python-two

工作队列（又称：任务队列）主要思想是避免立即执行资源密集型任务，并且必须等待它完成。相反，我们安排任务稍后完成。我们将任务封装为消息并将其发送到队列。在后台运行的工作进程将弹出任务并最终执行作业。当你运行许多工作程序时，它们之间将共享这些队列。

这个概念在Web应用程序中特别有用，在这些应用程序中，在短HTTP请求窗口期间无法处理复杂任务。

3.发布/订阅（Publish/Subscribe）
工作队列实际上就是将每个任务都交付给一个工作者（只有一个能接收）。在这一部分，我们将做一些完全不同的事情 – 我们将向多个消费者传递信息。此模式称为“发布/订阅”。

为了说明这种模式，我们将构建一个简单的日志记录系统。它将包含两个程序 – 第一个将发出日志消息，第二个将接收和打印它们。

在我们的日志记录系统中，接收程序的每个运行副本都将获取消息。这样我们就可以运行一个接收器并将日志定向到磁盘; 同时我们将能够运行另一个接收器并在屏幕上看到日志。

基本上，发布的日志消息将被广播给所有接收者。

Exchanges
我们向队列发送消息和从队列接收消息。现在是时候在RabbitMQ中引入完整的消息传递模型了。

RabbitMQ中消息传递模型的核心思想是生产者永远不会将任何消息直接发送到队列。实际上，生产者通常甚至不知道消息是否会被传递到任何队列。

相反，生产者只能向exchange发送消息。exchange是一件非常简单的事情。一方面，它接收来自生产者的消息，另一方面将它们推送到Queue。exchanges必须确切知道如何处理它收到的消息。它应该附加到特定队列吗？它应该附加到许多队列吗？或者它应该被丢弃。其规则由exchange类型定义 。

1556963392-3075-exchanges
有几种交换类型可供选择：direct，topic，headers 和fanout。我们将专注于最后一个 – fanout。让我们创建一个这种类型的exchange，并将其称为logs：

channel.exchangeDeclare(“logs”, “fanout”);
fanout exchange非常简单。正如您可能从名称中猜到的那样，它只是将收到的所有消息广播到它知道的所有队列中。而这正是我们记录器所需要的。

现在，我们可以发布到我们的命名交换：

channel.basicPublish（“logs”，""，null，message.getBytes（））;
临时队列
能够命名队列对我们来说至关重要 – 我们需要将工作人员指向同一个队列。当您想要在生产者和消费者之间共享队列时，为队列命名非常重要。

但我们的记录器并非如此。我们希望了解所有日志消息，而不仅仅是它们的一部分。我们也只对目前流动的消息感兴趣，而不是旧消息。要解决这个问题，我们需要两件事。

首先，每当我们连接到RabbitMQ时，我们都需要一个新的空队列。为此，我们可以使用随机名称创建队列，或者更好 – 让服务器为我们选择随机队列名称。

其次，一旦我们断开消费者，就应该自动删除队列。

在Java客户端中，当我们没有向queueDeclare()提供参数时，我们使用生成的名称创建一个非持久的，独占的并能勾自动删除队列：

String queueName = channel.queueDeclare().getQueue();
此时，queueName包含随机队列名称。例如：它可能看起来像amq.gen-JzTY20BRgKO-HjmUJj0wLg。

绑定队列
1556963392-9099-bindings
我们已经创建了一个fanout exchange和一个队列。现在我们需要告诉exchange将消息发送到我们的队列。exchange和队列之间的关系称为绑定。

channel.queueBind(queueName, “logs”, “”);
从现在开始，logs exchange会将消息附加到我们的队列中。

列出绑定
你可以使用rabbitmqctl列出现有的绑定

rabbitmqctl list_bindings
把它们放在一起
1556963393-7282-python-three-overall

以上的教材都来自官方文档的翻译，原文请查阅：

https://www.rabbitmq.com/getstarted.html

4.集成
maven

com.rabbitmq amqp-client 5.7.0 配置生产者 private static final String host = "127.0.0.1"; private static final int prot = 5672; /** * 缓存Chanel */ private Map

/**

• 连接工厂
  */
  private ConnectionFactory factory;

private Connection connection;
private Channel channel;

@Override
public void afterPropertiesSet() throws Exception {
factory = new ConnectionFactory();
factory.setHost(host);
factory.setPort(prot);
}

/**

• 获取一个连接，如果为空或断开了连接则重新实例化
• @return Connection
• @throws Exception
  */
  @Override
  public Connection getConnection() throws Exception {
  if (connection == null || !connection.isOpen()) {
  connection = factory.newConnection();
  }
  return connection;
  }

/**

• 返回一个通道
• @return Channel
• @throws Exception
  */
  @Override
  public Channel getChannel() throws Exception {
  if (channel == null || !channel.isOpen()) {
  channel = this.getConnection().createChannel();
  }
  return channel;
  }

/**

• 创建一个生产者，如果缓存中没有，则重新创建

• @param exchange Queue name|exchange name

• @param type queue|fanout|topic|headers|direct

• @param durable 是否持久性

• @return Channel

• @throws Exception
  */
  @Override
  public Channel createProducer(String exchange, String type, boolean durable) throws Exception {
  if (producerMap.containsKey(exchange + type + durable)) {
  logger(“producer by cache.”);
  Channel c1 = producerMap.get(exchange + type + durable);
  if (c1.isOpen()) {
  return c1;
  }
  }

  Channel c = this.getChannel();
  if (type == null || queue.equals(type)) {
  c.queueDeclare(exchange, durable, false, false, null);
  } else {
  c.exchangeDeclare(exchange, type, durable);
  }
  producerMap.put(exchange + type + durable, c);
  return c;
  }

/**

• 发送一条消息
• @param name Queue name|exchange name
• @param type queue|fanout|topic|headers|direct
• @param message content
• @return boolean
• @throws Exception
  */
  @Override
  public boolean send(String name, String type, String message) throws Exception {
  try {
  if (type == null || queue.equals(type)) {
  this.getProducer(name, type).basicPublish("", name, null, message.getBytes());
  } else {
  this.getProducer(name, type).basicPublish(name, “”, null, message.getBytes());
  }
  return true;
  } catch (Exception e) {
  e.printStackTrace();
  }
  return false;
  }
  配置消费者