RabbitMQ直接交换不消耗消息

Question

我有一个服务员线程想使用RabbitMQ direct exchange向Java中的客户线程发送一道寿司，但是我的客户没有收到这道菜。下面是我的服务员用来发布寿司菜肴对象的方法:

请注意，＜code＞dishKey＜/code＞作为参数传递，并在之前的if-else语句中被确定为＜code＞的“tamagoDishKey”＜/code>或＜code＞“ebiDishKey”的

public void waiterSendsDish (Sushi sushiDish, String dishKey) throws IOException, TimeoutException {
    
    String sushiDishEx = "sushiDishExchange";
    
    System.out.println("WAITER: Serving " + sushiDish.sushiName + " to Customer " + sushiDish.sushiNo + ".");
    
    ConnectionFactory factory = new ConnectionFactory();
    byte[] byteArray = getByteArraySushi(sushiDish);
    try (Connection con = factory.newConnection()){
        Channel chan = con.createChannel();
        chan.exchangeDeclare (sushiDishEx, BuiltinExchangeType.DIRECT);
        chan.basicPublish(sushiDishEx, dishKey, null, byteArray);
    }
}

以下是我的客户用来消费一种寿司（tamago）的方法：

public void tamagoReceiveDish() throws IOException, TimeoutException  {
String sushiDishEx = "sushiDishExchange";
String tamagoKey = "tamagoDishKey";
    
    ConnectionFactory factory = new ConnectionFactory();
    try (Connection con = factory.newConnection()) {  // added a try clause 
    Channel chan = con.createChannel();
   chan.exchangeDeclare(sushiDishEx, "direct"); // direct exchange
    
    String queueName = chan.queueDeclare().getQueue();  
    chan.queueBind(queueName, sushiDishEx, tamagoKey); 
    
    
    chan.basicConsume(queueName, true, (x, msg)->{
    byte[] byteArray = msg.getBody();
    try {
        Sushi sushi = (Sushi) deserialize(byteArray);
        System.out.println("CUSTOMER " + this.custNo + " (ordered Tamago Sushi): Received " + sushi.sushiName + ".");
        System.out.println("CUSTOMER " + this.custNo + ": Left the restaurant.");
        
    } catch (Exception e) {}
    }, x->{}
    );
    }
}

请注意，在为客户创建新连接时，我添加了一个< code>try子句,因为有许多客户(线程)被连续创建来订购和获取寿司，我希望客户只与服务员连接一次来获取寿司，然后停止连接(客户将离开餐厅)。

如果有人能帮我解决这个问题，我将不胜感激。我只想让一位独特的顾客接受每道菜，然后“离开餐厅”（关闭与该顾客的联系？）

Answer 1

我认为您无需等待消息即可关闭连接。没有什么能阻止你的代码关闭所有内容，这会杀死正在启动的消费者线程。

因此，您需要在一个time循环中等待：

private static final volatile boolean NO_LEFT = true;    
    public void tamagoReceiveDish() throws IOException, TimeoutException  {
    String sushiDishEx = "sushiDishExchange";
    String tamagoKey = "tamagoDishKey";
    
    ConnectionFactory factory = new ConnectionFactory();
    try (Connection con = factory.newConnection()) {  // added a try clause 
    Channel chan = con.createChannel();
   chan.exchangeDeclare(sushiDishEx, "direct"); // direct exchange
    
    String queueName = chan.queueDeclare().getQueue();  
    chan.queueBind(queueName, sushiDishEx, tamagoKey); 
    
    
    chan.basicConsume(queueName, true, (x, msg)->{
    byte[] byteArray = msg.getBody();
    try {
        Sushi sushi = (Sushi) deserialize(byteArray);
        System.out.println("CUSTOMER " + this.custNo + " (ordered Tamago Sushi): Received " + sushi.sushiName + ".");
        System.out.println("CUSTOMER " + this.custNo + ": Left the restaurant.");
        NO_LEFT = false;
        
    } catch (Exception e) {}
    }, x->{}
    );
   while(NO_LEFT){
     Thread.sleep(100);
    }
    }}