java-rabbitmq-官网实例04

java-rabbitmq-官网实例04


描述：
TOPIC 主题交换器的使用，模糊匹配路由键。


运行：
D4_EmitLogTopic.main();
D4_ReceiveLogsTopic.main();

package com.example.tutorials;


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


import java.util.Scanner;


/**
 *  topic"主题"交换器的"模糊匹配路由键"实例；
 * @create 2017-08-29
 * amqp-client 4.2.0
 **/
public class D4_EmitLogTopic {


    private static final String EXCHANGE_NAME = "topic_logs";


    public static void main(String[] argv) {
        Connection connection = null;
        Channel channel = null;
        try {
            ConnectionFactory factory = new ConnectionFactory();
            //设置登录账号
            factory.setHost(ServerInfo.host);
            factory.setPort(ServerInfo.port);
            factory.setUsername(ServerInfo.uName);
            factory.setPassword(ServerInfo.uPwd);
            //链接服务器
            connection = factory.newConnection();
            channel = connection.createChannel();
            //定义一个 topic 类型，交换器
            channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.TOPIC);
            //发送消息，使用指定路键
            System.out.println("输入要发送的消息，退出输入 x ");
            String message;
            String routingKey = "aaa.bbb.ccc";


            do{
                Scanner scanner = new Scanner(System.in);
                message = scanner.next();
                channel.basicPublish(EXCHANGE_NAME, routingKey, null, message.getBytes("UTF-8"));
                System.out.println(" [x] Sent '" + routingKey + "':'" + message + "'");
            }while(!"x".equals(message));
        }
        catch  (Exception e) {
            e.printStackTrace();
        }
        finally {
            if (connection != null) {
                try {
                    connection.close();
                }
                catch (Exception ignore) {}
            }
        }
    }
}

package com.example.tutorials;


import com.rabbitmq.client.*;


import java.io.IOException;
import java.util.concurrent.TimeoutException;


/**
 *  topic"主题"交换器的"模糊匹配路由键"实例；
 * amqp-client 4.2.0
 **/
public class D4_ReceiveLogsTopic {
    private static final String EXCHANGE_NAME = "topic_logs";


    public static void main(String[] argv) throws Exception {
        /*
        * 路由键占位符 #、* 作用
        * 星号：匹配一个词
        * 井号：匹配0到多个词
        * */
        stratThread("consumer_01","#");//可以收到消息(匹配任意路由)
        stratThread("consumer_02","aaa.*");//不能收到消息
        stratThread("consumer_03","*.bbb.*");//可以收到消息
        stratThread("consumer_04","*.bbb");//不能收到消息
        stratThread("consumer_05","aaa.*","*.bbb","*.*.*");//可以收到消息
    }


    public static void stratThread(final String consumerName, final String ...routingKeys){
        new Thread(new Runnable() {
            @Override
            public void run() {
                try {
                    startConsumer(consumerName,routingKeys);//启动消费者
                } catch (IOException e) {
                    e.printStackTrace();
                } catch (TimeoutException e) {
                    e.printStackTrace();
                }
            }
        }).start();
    }


    public static void startConsumer(final String consumerName,String ...routingKeys) throws IOException, TimeoutException {
        ConnectionFactory factory = new ConnectionFactory();
        //设置登录账号
        factory.setHost(ServerInfo.host);
        factory.setPort(ServerInfo.port);
        factory.setUsername(ServerInfo.uName);
        factory.setPassword(ServerInfo.uPwd);
        //链接服务器
        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();
        //定义一个 topic 类型，交换器
        channel.exchangeDeclare(EXCHANGE_NAME, "topic");
        String queueName = channel.queueDeclare().getQueue();


        //绑定队列到交换器,可以绑定多个路由键
        for (String bindingKey : routingKeys) {
            //使用指定的关键字，将队列绑定到一个交换,没有额外的参数。
            channel.queueBind(queueName, EXCHANGE_NAME, bindingKey);
        }


        System.out.println(" [*] Waiting for messages. To exit press CTRL+C");


        Consumer consumer = new DefaultConsumer(channel) {
            @Override
            public void handleConsumeOk(String consumerTag) {
                super.handleConsumeOk(consumerTag);
                System.out.println(" ["+consumerName+"] 已经注册成功！ ");
            }
            @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(" ["+consumerName+"] Received '" + envelope.getRoutingKey() + "':'" + message + "'");
            }
        };
        channel.basicConsume(queueName, true, consumer);
    }
}