java - 指定主题时使用直接交换的 RabbitMQ
问题描述
在我的应用程序中,我有 3 个类:
- 公司,它雇用工人来从事 3 种工作中的任何一种
- 工人,每个人可以做 2 份工作
- 管理员,它接收程序中所有消息的副本,并且可以向所有公司、所有工人或只是每个人
我work.companies.companyName
用于公司密钥和work.workers.workerName
工作人员密钥,它们都使用默认交换和队列进行通信。管理员通过admin
主题交换接收消息。
问题在于管理员 -> 其他所有人的沟通。它的工作原理与直接交换完全一样 - 我可以让公司和工人获得任何名称,甚至像“#”、“company1.#”等,他们不会收到任何东西,除非在管理员中我明确地使用“ work.companies.company1”。
我希望能够使用例如“work.companies.#”向所有公司发送消息。我究竟做错了什么?
管理员.java:
public class Administrator
{
public static void main(String[] args) throws IOException, TimeoutException
{
new Thread(new TopicListener("admin", ign -> {})).start();
TopicWriter writer = new TopicWriter();
// lots of code
主题监听器.java:
public class TopicListener implements Runnable
{
private final String EXCHANGE_NAME = "space";
private String key;
private Consumer<String> msgHandler;
public TopicListener(String key, Consumer<String> msgHandler)
{
this.key = key;
this.msgHandler = msgHandler;
}
@Override
public void run()
{
try
{
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("localhost");
Connection connection = factory.newConnection();
Channel channel = connection.createChannel();
channel.exchangeDeclare(EXCHANGE_NAME, "topic");
String queueName = channel.queueDeclare().getQueue();
channel.queueBind(queueName, EXCHANGE_NAME, key);
com.rabbitmq.client.Consumer consumer = new DefaultConsumer(channel)
{
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body)
{
String msg = new String(body, StandardCharsets.UTF_8);
System.out.println("Received: \"" + msg + "\"");
msgHandler.accept(msg);
}
};
channel.basicConsume(queueName, true, consumer);
}
catch (IOException | TimeoutException e)
{ e.printStackTrace(); }
}
}
TopicWriter.java:
public class TopicWriter
{
private final String EXCHANGE_NAME = "space";
private final Channel channel;
public TopicWriter() throws IOException, TimeoutException
{
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("localhost");
Connection connection = factory.newConnection();
this.channel = connection.createChannel();
channel.exchangeDeclare(EXCHANGE_NAME, "topic");
}
public void send(String msg, String key) throws IOException
{
channel.basicPublish(
EXCHANGE_NAME,
key,
null,
msg.getBytes(StandardCharsets.UTF_8));
}
}
Company.java 包含:
new Thread(new TopicListener("space.agencies." + agencyID, ign -> {})).start();
Worker.java 包含:
new Thread(new TopicListener("space.carriers." + carrierID, consumer)).start();
解决方案
我发现问题出在哪里:我试图使用 Topic 向每个人发送消息,在 RabbitMQ 中,Topic 用于指定谁应该接收消息。“#”或“*”应该在队列键声明中使用,而不是在使用给定键发送消息时使用。
推荐阅读
- java - 如何更改 hpi:run 的端口号
- docker - 如何在 CircleCI 2.0 的部署步骤中使用构建步骤中的 docker 映像?
- php - 如何将base64图像解码保存到laravel中的公用文件夹
- prism - PRISM + Splat 作为 IOC
- html - Flexbox - 如何在导航栏中对齐项目
- javascript - 为什么 app.get 中间件在 expressjs 中的 app.use 中间件之后不起作用
- php - 如何使用 php 中的 if else 语句将每隔一个数字替换为另一个数字?
- c# - C# WPF MVVM - 将多个 csv 合并到数据网格
- c - What is int (*fp)()
- javascript - 如何在没有控制器的情况下访问视图中的 dataModel