美文网首页
RabbitMQ五种模式(TOPICS模式)

RabbitMQ五种模式(TOPICS模式)

作者: OomelodyoO | 来源:发表于2018-09-03 11:05 被阅读0次

    Send.java

    public class Send {
        public static String EXCHANGE_NAME = "test_exchange_topic";
    
        public static void main(String[] args) throws IOException, TimeoutException {
            ConnectionFactory factory = new ConnectionFactory();
            factory.setHost("localhost");
            factory.setVirtualHost("/zhang");
            factory.setUsername("zhang");
            factory.setPassword("zhang");
            Connection connection = factory.newConnection();
            Channel channel = connection.createChannel();
            channel.exchangeDeclare(EXCHANGE_NAME, "topic");
            String message = "message";
            channel.basicPublish(EXCHANGE_NAME, "item.broadcast", null, message.getBytes());
            channel.close();
            connection.close();
        }
    }
    

    Recv.java

    public class Recv {
        public static String QUEUE_NAME = "test_queue_topic_1";
        public static String EXCHANGE_NAME = "test_exchange_topic";
    
        public static void main(String[] args) throws IOException, TimeoutException {
            ConnectionFactory factory = new ConnectionFactory();
            factory.setHost("localhost");
            factory.setVirtualHost("/zhang");
            factory.setUsername("zhang");
            factory.setPassword("zhang");
            Connection connection = factory.newConnection();
            Channel channel = connection.createChannel();
            channel.queueDeclare(
                    QUEUE_NAME, false, false,
                    false, null
            );
            channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "item.broadcast");
            channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "item.a");
            channel.basicQos(1);
            DefaultConsumer 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 '" + message + "'");
                    try {
                        Thread.sleep(2 * 1000);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    channel.basicAck(envelope.getDeliveryTag(), false);
                }
            };
            channel.basicConsume(QUEUE_NAME, false, consumer);
        }
    }
    

    Recv2.java

    public class Recv2 {
        public static String QUEUE_NAME = "test_queue_topic_2";
        public static String EXCHANGE_NAME = "test_exchange_topic";
    
        public static void main(String[] args) throws IOException, TimeoutException {
            ConnectionFactory factory = new ConnectionFactory();
            factory.setHost("localhost");
            factory.setVirtualHost("/zhang");
            factory.setUsername("zhang");
            factory.setPassword("zhang");
            Connection connection = factory.newConnection();
            Channel channel = connection.createChannel();
            channel.queueDeclare(
                    QUEUE_NAME, false, false,
                    false, null
            );
            channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "item.broadcast");
            channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "item.b");
            channel.basicQos(1);
            DefaultConsumer 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 '" + message + "'");
                    try {
                        Thread.sleep(2 * 1000);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    channel.basicAck(envelope.getDeliveryTag(), false);
                }
            };
            channel.basicConsume(QUEUE_NAME, false, consumer);
        }
    }
    

    build.gradle

    plugins {
        id 'java'
    }
    
    version = '0.0.1-SNAPSHOT'
    sourceCompatibility = 1.8
    
    repositories {
        mavenCentral()
    }
    
    dependencies {
        compile group: 'com.rabbitmq', name: 'amqp-client', version: '5.3.0'
        testCompile group: 'junit', name: 'junit', version: '4.12'
    }
    

    rabbit-mq源码地址​github.com

    相关文章

      网友评论

          本文标题:RabbitMQ五种模式(TOPICS模式)

          本文链接:https://www.haomeiwen.com/subject/awdbwftx.html