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

2018-11-21 本文已影响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();
    }
}

执行上面代码之后，结果如下：