用rabbitMQ實現消息的發佈和訂閱

RabbitMQ能夠實現消息發佈和訂閱，這其中須要exchange，消息的發佈者，消費者，消息隊列和Exchange關係以下圖

消息的生產者發出消息，而後通過Exchange轉換，消息隊列（queue）須要綁定Exchange，Exchange把消息發送到各個消息隊列中，而後，各個消費者從消息隊列中取到發佈者發佈的消息。

利用Java做爲客戶端，具體代碼以下：

發佈者代碼

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;

/**
 *
 * 用rabbitMQ實現發佈訂閱
 * 發佈類
 * Created by wangtf on 2015/11/17.
 */
public class Emitlog {

    private static final String EXCHANGE_NAME = "logs";

    public  static  void main(String args[]) throws  Exception{
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost("localhost");
        Connection connection = connectionFactory.newConnection();
        Channel channel = connection.createChannel();
        channel.exchangeDeclare(EXCHANGE_NAME,"fanout");
        String msg = "各單位請注意";
        channel.basicPublish(EXCHANGE_NAME,"",null,msg.getBytes());
        System.out.println("[send] msg: " +msg);
        channel.close();
        connection.close();
    }
}

消息接受類

import com.rabbitmq.client.*;

import java.io.IOException;

/**
 * RabbitMQ實現發佈訂閱的功能
 * 訂閱類
 * Created by wangtf on 2015/11/17.
 */
public class ReceiveLogs {

    private static final String EXCHANGE_NAME = "logs";
    public static void main(String[] args) throws  Exception{
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();
        channel.exchangeDeclare(EXCHANGE_NAME,"fanout");
        String queueName = channel.queueDeclare().getQueue();
        channel.queueBind(queueName,EXCHANGE_NAME,"");
        System.out.print("[*] waiting for message");

        DefaultConsumer consume = 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("[x] receive message :" + message);
            }
        };

        channel.basicConsume(queueName,true,consume);
    }
}

分別運行ReceiveLogs和Emitlog，能獲得以下結果：

【Emitlog】
[send] msg: 各單位請注意
Process finished with exit code 0

【ReceiveLogs】
[*] waiting for message[x] receive message :各單位請注意