RabbitMQ入门_07_Fanout 与 Topic

kevin0490

public>　　private static final String EXCHANGE_NAME = "StatusUpdateFanout";
　　public static void main(String[] argv) throws Exception {
　　ConnectionFactory factory = new ConnectionFactory();
　　factory.setHost("localhost");
　　Connection connection = factory.newConnection();
　　final Channel senderChannel = connection.createChannel();
　　senderChannel.exchangeDeclare(EXCHANGE_NAME, "fanout");
　　final CountDownLatch latch = new CountDownLatch(1);
　　new Thread(new Runnable() {
　　@Override
　　public void run() {
　　try {
　　for (int i = 0; i < 10;) {
　　String message = &quot;NO. &quot; + ++i;
　　TimeUnit.MILLISECONDS.sleep(100);
　　senderChannel.basicPublish(EXCHANGE_NAME, &quot;&quot;, null, message.getBytes(&quot;UTF-8&quot;));
　　System.out.printf(&quot;(%1$s)[===>%2$s    ] %3$s\n&quot;, &quot;S&quot;, EXCHANGE_NAME + &quot;:&quot;, message);
　　if (i == 4) {
　　latch.countDown();
　　}
　　}
　　senderChannel.close();
　　} catch (Exception e) {
　　}
　　}
　　}).start();
　　final Channel consumerChannel1 = connection.createChannel();
　　consumerChannel1.queueDeclare(&quot;SystemA&quot;, false, false, true, null);
　　consumerChannel1.queueBind(&quot;SystemA&quot;, EXCHANGE_NAME, &quot;&quot;);
　　consumerChannel1.basicQos(3);
　　Consumer consumer1 = new DefaultConsumer(consumerChannel1) {
　　@Override
　　public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body)
　　throws IOException {
　　String message = new String(body, &quot;UTF-8&quot;);
　　System.out.printf(&quot; [    %2$s<===](%1$s) %3$s\n&quot;, &quot;A&quot;, &quot;SystemA&quot;, message);
　　try {
　　TimeUnit.MILLISECONDS.sleep(200);
　　} catch (InterruptedException e) {
　　}
　　consumerChannel1.basicAck(envelope.getDeliveryTag(), false);
　　}
　　};
　　consumerChannel1.basicConsume(&quot;SystemA&quot;, false, consumer1);
　　latch.await();
　　final Channel consumerChannel2 = connection.createChannel();
　　consumerChannel2.queueDeclare(&quot;SystemB&quot;, false, false, true, null);
　　consumerChannel2.queueBind(&quot;SystemB&quot;, EXCHANGE_NAME, &quot;&quot;);
　　consumerChannel2.basicQos(3);
　　Consumer consumer2 = new DefaultConsumer(consumerChannel2) {
　　@Override
　　public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body)
　　throws IOException {
　　String message = new String(body, &quot;UTF-8&quot;);
　　System.out.printf(&quot; [    %2$s<===](%1$s) %3$s\n&quot;, &quot;B -- won't receive first 4 messages&quot;, &quot;SystemB&quot;, message);
　　try {
　　TimeUnit.MILLISECONDS.sleep(200);
　　} catch (InterruptedException e) {
　　}
　　consumerChannel2.basicAck(envelope.getDeliveryTag(), false);
　　}
　　};
　　consumerChannel2.basicConsume(&quot;SystemB&quot;, false, consumer2);
　　}
　　}