RabbitMQ入門-Topic模式 - bigdataZJ－IT工程師數位筆記本

文章出處

上篇《RabbitMQ入門-Routing直連模式》我們介紹了可以定向發送消息，并可以根據自定義規則派發消息。看起來，這個Routing模式已經算靈活的了，但是，這還不夠，我們還有更加多樣靈活的Topic模式。

Topic模式

模型組成相較前幾種沒有什么變化，一個生產者P，一個交換機X，多個消息隊列Q以及多個消費者C
在Exchange派發消息到消息隊列Queue所用的規則不同，我們看到了有符號"*"以及"#"，可以認為是通配符
"*"用于匹配一個單詞，比如"a","abc"等；"#"用于匹配0個或者多個單詞，比如"", "abc", "abc.def"等

發送端

/**
 * Created by jackie on 17/8/7.
 */
public class EmitLogDirect {

    private static final String EXCHANGE_NAME = "direct_logs";

    public static void main(String[] argv) throws Exception {

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

        channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.DIRECT);

        String severity = getSeverity(argv);
        String message = getMessage(argv);

        channel.basicPublish(EXCHANGE_NAME, severity, null, message.getBytes("UTF-8"));
        System.out.println(" [x] Sent '" + severity + "':'" + message + "'");

        channel.close();
        connection.close();
    }

    private static String getSeverity(String[] strings){
        if (strings.length < 1)
            return "info";
        return strings[0];
    }

    private static String getMessage(String[] strings){
        if (strings.length < 2)
            return "Hello World!";
        return joinStrings(strings, " ", 1);
    }

    private static String joinStrings(String[] strings, String delimiter, int startIndex) {
        int length = strings.length;
        if (length == 0 ) return "";
        if (length < startIndex ) return "";
        StringBuilder words = new StringBuilder(strings[startIndex]);
        for (int i = startIndex + 1; i < length; i++) {
            words.append(delimiter).append(strings[i]);
        }
        return words.toString();
    }
}

channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.TOPIC);這里指定的Exchagne模式為Topic模式
通過String routingKey = getRouting(argv)；實現在Program arguments中填寫routing key參數
通過String message = getMessage(argv)；實現在Program arguments中填寫發送的消息

這時候我們給Program argument賦值如下，并啟動發送端程序

程序運行完，可以在RabbitMQ管理應用中看到名為“topic_logs”的Exchange。

接收端

/**
 * Created by jackie on 17/8/7.
 */
public class ReceiveLogsDirect {

    private static final String EXCHANGE_NAME = "direct_logs";

    public static void main(String[] argv) throws Exception {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("192.168.3.161");
        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();

        channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.DIRECT);
        String queueName = channel.queueDeclare().getQueue();

        if (argv.length < 1){
            System.err.println("Usage: ReceiveLogsDirect [info] [warning] [error]");
            System.exit(1);
        }

        for(String severity : argv){
            channel.queueBind(queueName, EXCHANGE_NAME, severity);
        }
        System.out.println(" [*] Waiting for messages. To exit press CTRL+C");

        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(" [x] Received '" + envelope.getRoutingKey() + "':'" + message + "'");
            }
        };
        channel.basicConsume(queueName, true, consumer);
    }
}