Producer端

1、channel的建立

　　無論是才用什麼樣的Exchange，建立channel程式碼都是相同的，如下  

1 ConnectionFactory factory = new ConnectionFactory();
2 factory.setHost("localhost");
3 Connection connection = factory.newConnection();
4 Channel channel = connection.createChannel();

2、Exchange的建立

　2.1 direct

　　direct使用預設的Exchange，不需要宣告，單需要指定訊息傳送到那個佇列

channel.queueDeclare(QUEUE_NAME, false, false, false, null);

　2.2 fanout

channel.exchangeDeclare(EXCHANGE_NAME, "fanout")

　2.3 topic如下

channel.exchangeDeclare(EXCHANGE_NAME, "topic");

3、訊息的傳送

  3.1 direct

channel.basicPublish("", QUEUE_NAME, null
 
, message.getBytes());

  3.2 fanout 

channel.basicPublish(EXCHANGE_NAME, "", null, message.getBytes());

  3.3 topic

channel.basicPublish(EXCHANGE_NAME, routingKey, null, message.getBytes());

cusumer端

1、建立channel 

ConnectionFactory factory = new
 
 ConnectionFactory();
factory.setHost("localhost");
Connection connection = factory.newConnection();
Channel channel = connection.createChannel();

2、訊息前的準備

   2.1  direct直接繫結佇列進行訊息的消費        

chanel.queueDeclare(QUEUE_NAME, false, false, false, null);

   2.2 fanout，需要先指定exchange型別為fanout 

channel.exchangeDeclare(EXCHANGE_NAME, "fanout");
String queueName = channel.queueDeclare().getQueue();
channel.queueBind(queueName, EXCHANGE_NAME, "");

  2.3 topic       

channel.exchangeDeclare(EXCHANGE_NAME, "topic");
String queueName = channel.queueDeclare().getQueue();
 for(String bindingKey : argv){
      channel.queueBind(queueName, EXCHANGE_NAME, bindingKey);
 }

3、具體消費訊息的程式碼是一樣的

QueueingConsumer consumer = new QueueingConsumer(channel);
channel.basicConsume(QUEUE_NAME, true, consumer);
while (true) {
    QueueingConsumer.Delivery delivery = consumer.nextDelivery();
    Envelope elp = delivery.getEnvelope();
    String message = new String(delivery.getBody());
    System.out.println(" [x] Received '" + message + "'");
    channel.basicAck(elp.getDeliveryTag(), false);
    //channel.basicNack();
​}