RabbitMQコンシューマは、送信されたメッセージの一部のみを消費します

Question

私はRabbitMQに少数のメッセージを送信しますが、コンシューマは2つしか消費しません。

@Component 
public class Producer implements CommandLineRunner { 

private final RabbitTemplate rabbitTemplate; 
private final Consumer receiver; 
private final ConfigurableApplicationContext context; 

public Producer(Consumer receiver, RabbitTemplate rabbitTemplate, 
     ConfigurableApplicationContext context) { 
    this.receiver = receiver; 
    this.rabbitTemplate = rabbitTemplate; 
    this.context = context; 
} 

private List<Message> extractphoneNumberNumbers(String path){ 
    List<Message> messages = new ArrayList(); 
    try{ 
     BufferedReader fileReader = new BufferedReader(
        new InputStreamReader(new ClassPathResource(path).getInputStream())); 
     JsonReader reader = new JsonReader(fileReader); 
     Gson gson = new GsonBuilder().create(); 
     messages = new Gson().fromJson(reader, new TypeToken<List<Message>>() {}.getType()); 
    }catch(Exception e){ 
     System.out.println(e.getMessage()); 
    } 
    return messages; 
} 

public void run(String... args) throws Exception { 
    List<Message> messagesToPublish = extractphoneNumberNumbers("phoneNumbers"); 
    AtomicInteger messageId = new AtomicInteger(); 
    for(Message message : messagesToPublish){ 
     System.out.println("Sent " + message.toString()); 
     rabbitTemplate.convertAndSend(App.queueName, new Message(messageId.incrementAndGet(), message.getPhoneNumber())); 
    } 
    receiver.getLatch().await(10000, TimeUnit.MILLISECONDS); 

    context.close(); 
} 

} 


@Component 
public class Consumer { 
private CountDownLatch latch = new CountDownLatch(1); 
private HashMap<String, Integer> phoneCountryCode = new HashMap<String, Integer>(); 
private int i = 0; 

public void receiveJsonMessages(Message message){ 
    System.out.println("Received " + message.toString()); 
    phoneCountryCode.put(message.getCountryCode(), phoneCountryCode.getOrDefault(message.getCountryCode(), 0) + 1); 

    for(Map.Entry<String, Integer> entry : phoneCountryCode.entrySet()){ 
     System.out.println(entry.getKey() + " " + entry.getValue()); 
    } 

    latch.countDown(); 
} 

public CountDownLatch getLatch() { 
    return latch; 
} 
} 

---

Received 5 - +40722579707 - RO 
Sent 0 - +447984627687 - GB 
RO 1 
Received 6 - +40722579717 - RO 
Sent 0 - +447984627657 - GB 
RO 2 
Sent 0 - +447984627667 - GB 
Sent 0 - +447984627677 - GB 
Sent 0 - +40722579707 - RO 
Sent 0 - +40722579717 - RO 
Sent 0 - +40722579727 - RO 

目標はグループにあり、国によって電話番号を数える：私のプロデューサー&消費者のためのコードの下にして下さい。私はちょうどその時点でRabbitMQで遊んでいて、経験がないので、グループ化は...今はばかげている。

なぜ、7ではなく2つのメッセージしか受信できないのですか？

編集：私は、労働者（生産者）が完成すると豆が止まるという事実によると信じています。コンシューマーが終了した後、これをどのように停止するように設定できますか？

Answer 1

ここでは、https://www.rabbitmq.com/tutorials/tutorial-one-spring-amqp.htmlのサンプルに従います。重要な部分は

@RabbitHandler 
public void receive(String in) { 
    System.out.println(" [x] Received '" + in + "'"); 
} 

方法は、シーンの背後にあるループ内で実行され、それが利用可能な今までにときにメッセージを消費しています。上記のようなコンテキストを本当に閉じるには、消費されるメッセージの数を知り、必要な数のメッセージでCountDownLatchを設定する必要があります。これまでにメッセージを受信すると、カウントダウンしてメインスレッドのラッチを待つことができます。しかし、これはすべてのメッセージを受け取ることを保証するものではなく、ネットワーク障害など何かが起きる可能性があるという保証はありません。その後、ラッチ待機時にタイムアウトを設定することができます。