Rabbitmq - 新消费者没有收到任何东西

Question

我使用rabbitMQ 来将任务发送给在运行时创建的工作人员（消费者）。目前，每次创建新任务时，都会创建一个新的工作线程。问题是这样的： - 用户创建一个任务

-创建一个worker，然后将任务发送到队列上供worker处理

-worker开始处理队列（worker基本上休眠了一段时间）

-另一个用户创建任务

-创建新的工作人员并将任务发送到队列上

-新的工作人员不处理新任务，同时绝对不执行任何操作，并且新任务由第一个工作人员在完成第一个任务后处理

我检查了rabbitmq的管理部分，有两个消费者绑定到队列，但其中一个似乎完成了所有工作，而另一个只是等待。

这是工人的代码： public class Worker extends Thread {

private final static String QUEUE_NAME = "Tasks";
private final static String QUEUE_COMPL = "Completed";
public static int id = 0;
private static final String EXCHANGE_NAME = "logs";
public int compteur;
String identifier;

public Worker() {
    Worker.id++;
    compteur = id;
}
public void run() {
    try {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        channel.basicQos(1);
        final Consumer consumer = new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                String message = new String(body, "UTF-8");
                System.out.println("WORKER " + compteur + " : Message received :" + message);

                String taskName = message.split(" ")[0];
                String length = message.split(" ")[1];

                try {
                    System.out.println("WORKER " + compteur + " : Commencing job :" + message);
                    doWork(length);
                    System.out.println("WORKER " + compteur + " : Job's finished :" + message);
                    taskName += " done by " + compteur;
                   // confirm(taskName);
                } catch (InterruptedException e) {
                    // TODO Auto-generated catch block
                    e.printStackTrace();
                } finally {
                    System.out.println("WORKER " + compteur + " : Waiting for a new task...");
                }

            }
        };

        channel.basicConsume(QUEUE_NAME, true, consumer);
    } catch (IOException ex) {
            Logger.getLogger(Worker.class.getName()).log(Level.SEVERE, null, ex);
    } catch (TimeoutException ex) {
            Logger.getLogger(Worker.class.getName()).log(Level.SEVERE, null, ex);
    }
}

private static void doWork(String taskLength) throws InterruptedException {
    int temps = Integer.parseInt(taskLength);
    Thread.sleep(temps);
}
}

以及将消息放入队列的部分的代码： public class serverSD {

private final static String QUEUE_NAME = "Tasks";
private  Manager MANAGER = new Manager();

@WebMethod(operationName = "processTask")
public String processTask(@WebParam(name = "message") String txt, @WebParam(name = "duree") int duree) throws IOException, TimeoutException {
    if (MANAGER == null){
        MANAGER= new Manager();
        MANAGER.listen();
    }
    System.out.println("SERVER : Message received : " + txt + " " + duree);
    MANAGER.giveTask();
    ConnectionFactory factory = new ConnectionFactory();
    String message = txt + " " + duree;
    System.out.println("SERVER : Sending message to workers : " + message);
    factory.setHost("localhost");
    Connection connection = factory.newConnection();
    Channel channel = connection.createChannel();
    channel.queueDeclare(QUEUE_NAME, false, false, false, null);

    channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
    channel.close();
    connection.close();       
    return "Your task is being processed";
}
}

（Manager 是创建工作人员的类。）

如果已经有人问过类似的问题，但我找不到它，我很抱歉。感谢您提供任何可能的帮助:)

Answer 1

该方法的第二个参数basicConsume是“自动确认”。将此参数设置为true意味着消费者一旦收到消息就会告诉 RabbitMQ 该消息已被确认。

当消费者设置为 autoAck true 时，即使设置为 1，它也很有可能立即从队列中接收下一条可用消息。basicQos这种情况会发生，因为消费者会立即递减 1 限制，也就是说不再有任何消息，它可以接受下一条消息。

与基本 QoS 设置 1 结合使用时，更改自动确认参数可以false防止此问题，因为它会迫使您的消费者说“嘿，我收到一条消息，目前正在处理它。不要发送给我”在我做完之前还有别的事。”

这使得其他消费者可以说“嘿，我有空位。请继续向我发送消息”