美文网首页
RabbitMQ系列 通过Rabbit-client操作mq(e

RabbitMQ系列 通过Rabbit-client操作mq(e

作者: 时之令 | 来源:发表于2018-11-21 18:54 被阅读0次
    1,导入依赖包pom.xml
    <project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
        xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
        <modelVersion>4.0.0</modelVersion>
    
        <groupId>com.edu.mq</groupId>
        <artifactId>rabbitmq-java-client</artifactId>
        <version>0.0.1-SNAPSHOT</version>
        <packaging>jar</packaging>
    
        <name>rabbitmq-java-client</name>
        <url>http://maven.apache.org</url>
    
        <properties>
            <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
            <maven.compiler.source>1.8</maven.compiler.source>
            <maven.compiler.target>1.8</maven.compiler.target>
        </properties>
    
        <dependencies>
            <dependency>
                <groupId>com.rabbitmq</groupId>
                <artifactId>amqp-client</artifactId>
                <version>4.0.2</version>
            </dependency>
        </dependencies>
    </project>
    
    

    2,通过rabbitClient创建一个新的exchange

     public static void main(String[] args) throws IOException, TimeoutException {
    
            //常见连接工厂
            ConnectionFactory factory = new ConnectionFactory();
            factory.setHost("localhost");
            factory.setPort(5672);
            factory.setVirtualHost("/");
            factory.setUsername("guest");
            factory.setPassword("guest");
    //上述创建工厂的时候,配置factory信息比较繁琐,有多种方式
    //可以配置。
    
            //获取连接
            Connection connection = factory.newConnection();
    
            //获取channel,所有的操作都是在channel,一个connection可以创建多个channel
            Channel channel = connection.createChannel();
    
            channel.exchangeDeclare("logs","direct");
    
            channel.close();
            connection.close();
        }
    

    结果如图:


    image.png

    上图的logs 是一个没有任何features的exchange。

    通过rabbitMQ的channel,exchange,queue,binding的创建,删除,属性等操作
    package com.edu.mq.rabbitmq;
    
    import com.rabbitmq.client.*;
    
    import java.io.IOException;
    import java.util.HashMap;
    import java.util.Map;
    import java.util.concurrent.TimeoutException;
    
    /**
     * 类说明:
     *
     * @author zhangkewei
     * @date 2018/11/21下午4:40
     */
    public class App {
    
        public static void main(String[] args) throws IOException, TimeoutException, InterruptedException {
    
            //常见连接工厂
            ConnectionFactory factory = new ConnectionFactory();
            factory.setHost("localhost");
            factory.setPort(5672);
            factory.setVirtualHost("/");
            factory.setUsername("guest");
            factory.setPassword("guest");
    
            //获取连接
            Connection connection = factory.newConnection();
    
            //获取channel,所有的操作都是在channel,一个connection可以创建多个channel
            Channel channel = connection.createChannel();
    
    
            /*****************************************  exchange开始   ********************************************/
            /**
             * exchange的一些操作,创建,删除,是否存在,添加是否持久化,自动删除,类型,alternate-exchange,其中重复创建
             * 已经存在的exchange是不会有问题的,当重复创建的时候,如果已经存在,不会在创建
             */
            channel.exchangeDeclare("logs", "direct");
            //通过枚举的方式写type,防止自己写的时候,拼写错误。
            channel.exchangeDeclare("log.info.first", BuiltinExchangeType.DIRECT);
            //其中的第三个参数表示是否持久化。
            AMQP.Exchange.DeclareOk declareOk = channel.exchangeDeclare("log.info.two", BuiltinExchangeType.DIRECT, true);
    
            //其中的第四个参数表示是否自动删除。第五个参数表示是否给exchange 添加额外的exchange,当没有路由到队列的时候,消息转发到log的exchange中。
            Map<String, Object> arguments = new HashMap();
            arguments.put("alternate-exchange", "log");
            channel.exchangeDeclare("log.warn", BuiltinExchangeType.DIRECT, true, false, arguments);
    
            //判断某个exchange是否存在。
            channel.exchangeDeclare("log", BuiltinExchangeType.DIRECT);
            AMQP.Exchange.DeclareOk log = channel.exchangeDeclarePassive("log");
            System.out.println(log);
            //如果没有会报错
            // log = channel.exchangeDeclarePassive("log2");
            // System.out.println(log);
    
            // 删除某个exchange,如果没有这个这个exchange会报错
            AMQP.Exchange.DeleteOk log1 = channel.exchangeDelete("log");
            /****************************************    exchange结束    *******************************************/
    
    /*———————————————————————————————————————————————————————————————————————————————————————————————————————————————————————————*/
            /*****************************************  queue开始   ********************************************/
    
            //创建队列(可以重复执行).如果已经有该队列,而且该队列有消息,这个时候创建也没有问题,不会删除原有的队列中的消息
    
            //其中第一个参数是队列名称,第二个是是否持久化,第三个参数是是否是排他的(ture的时候,如果connection断开,就会消失)
            //第四个参数表示是否持久化,第五个是队列的一些属性的设置,可以通过map设置
            channel.queueDeclare("debug_queue", true, false, false, null);
            //队列是否存在,如果不存在会报错。
            channel.queueDeclarePassive("debug_queue");
            /*****************************************  queue结束   ********************************************/
    
     /*———————————————————————————————————————————————————————————————————————————————————————————————————————————————————————————*/
            /*****************************************  binding开始   ********************************************/
            //绑定分为两种,1,exchange和queue绑定。2,exchange和exchange绑定。都可以重复执行
            //将队列绑定到exchange中,第一个参数是队列名称,第二个参数表示exchange名称,第三个参数表示routingKey
            channel.queueBind("debug_queue","log.warn","info");
            channel.exchangeBind("log.warn","log.info","log");
    
            //逆向操作,解绑,可以重复执行
           // channel.queueUnbind("debug_queue","log.warn","info");
           // channel.exchangeUnbind("log.warn","log.info","log");
            /*****************************************  binding结束   ********************************************/
    
            /**
             * 以上的exchange,queue,binding都有异步的操作XXXNoWait方法,不等待执行结果。
             */
    
            Thread.sleep(5000L);
            channel.close();
            connection.close();
        }
    }
    
    

    执行上面代码之后,结果如下:


    创建的exchange信息 创建的队列信息 队列和exchange绑定 exchange和exchange绑定

    提示:有些说明在代码中。

    相关文章

      网友评论

          本文标题:RabbitMQ系列 通过Rabbit-client操作mq(e

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