码迷，mamicode.com
首页 > 编程语言 > 详细

喵星之旅-狂奔的兔子-rabbitmq的java客户端使用入门

时间：2020-02-27 20:37:55      阅读：67      评论：0      收藏：0      [点我收藏+]

标签：工具   添加   代码   etc   单词   快速   有一个   针对   code   

一、简介

RabbitMQ是实现了高级消息队列协议（AMQP）的开源消息代理软件（亦称面向消息的中间件）。

消息队列都涉及的生产者消费者模型，不做详解，本文只作为快速使用的参考文档。

消息队列主要有点对点和发布订阅模式。

其主要用途是异步、削峰，充当一个缓存的作用。只有可以异步处理时才可以使用消息队列。

官方参考文档地址：

https://www.rabbitmq.com/getstarted.html

需要添加jar包：

amqp-client-3.5.6.jar hamcrest-core-1.3.jar

本文jdk版本1.8。


二、直接使用队列

1、简单队列

特点是生产者消费者一一对应。

生产者一般效率很高，生产者很低，所以一一对应是不对的。所以这个一般不使用。

其结构如图所示：

技术图片

 

先创建一个工具类：

package com.bunny.rabbit;
/**
 * 
 * @author bunny～～我是兔子我会喵，我叫喵星兔。
 * 连接工具类
 */

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

public class LianjieGongjulei {
    public static final String SIMPLE_QUEUE_NAME = "simplequeue";
    public static final String WORK_QUEUE_NAME = "workqueue";
    public static final String PUBSUB_EXCHANGE_NAME = "pubsubexchange";
    public static final String ROUTING_EXCHANGE_NAME = "routingexchange";
    public static final String TOPIC_EXCHANGE_NAME = "topicexchange";
    public static final String PUBSUB_QUEUE_NAME1 = "pubsubqueue1";
    public static final String PUBSUB_QUEUE_NAME2 = "pubsubqueue2";
    public static final String ROUTING_QUEUE_NAME1 = "routingqueue1";
    public static final String ROUTING_QUEUE_NAME2 = "routingqueue2";
    public static final String TOPIC_QUEUE_NAME1 = "topicqueue1";
    public static final String TOPIC_QUEUE_NAME2 = "topicqueue2";
    
    public static Connection getConnection() {
        try {
            ConnectionFactory cf = new ConnectionFactory();
            cf.setHost("127.0.0.1");
            cf.setPort(5672);
            cf.setVirtualHost("/mybunny");
            cf.setUsername("bunny");
            cf.setPassword("bunny");
            return cf.newConnection();
        } catch (Exception e) {
            e.printStackTrace();
        }
        return null;
    }

}

生产者：

package com.bunny.rabbit.simple;

import java.io.IOException;

import com.bunny.rabbit.LianjieGongjulei;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;

/**
 * 
 * @author bunny～～我是兔子我会喵，我叫喵星兔。
 *  简单队列生产者
 */
public class Shengchanzhe {
    public static void main(String[] args) throws Exception {
        Connection c = LianjieGongjulei.getConnection();
        Channel ch = c.createChannel();
        //如果已经存在，可以不声明，声明的话也没错。所以不管是否存在都可以声明。
        ch.queueDeclare(LianjieGongjulei.SIMPLE_QUEUE_NAME, false, false, false, null);

        String mes = "Hello Kitty!";
        ch.basicPublish("", LianjieGongjulei.SIMPLE_QUEUE_NAME, null, mes.getBytes());
        System.out.println(" bunny Sent ‘" + mes + "‘");

        ch.close();
        c.close();

    }
}

消费者：

package com.bunny.rabbit.simple;

import java.io.IOException;

import com.bunny.rabbit.LianjieGongjulei;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.QueueingConsumer;

/**
 * 
 * @author bunny～～我是兔子我会喵，我叫喵星兔。 
 * 简单队列消费者
 */
public class Xiaofeizhe {
    public static void main(String[] args) throws Exception {
        Connection c = LianjieGongjulei.getConnection();
        Channel ch = c.createChannel();
        //如果已经存在，可以不声明，声明的话也没错。所以不管是否存在都可以声明。
        ch.queueDeclare(LianjieGongjulei.SIMPLE_QUEUE_NAME, false, false, false, null);

        // 定义消费者
        QueueingConsumer consumer = new QueueingConsumer(ch);
        ch.basicConsume(LianjieGongjulei.SIMPLE_QUEUE_NAME, true, consumer);

        // 获取消息
        while (true) {
            QueueingConsumer.Delivery delivery = consumer.nextDelivery();
            String message = new String(delivery.getBody());
            System.out.println(" bunny Received ‘" + message + "‘");
        }

    }
}

 

2、work队列

手工确认则会按照消费端的能力进行分发（公平分发），否则是按照生产者的规则分发（轮询）。

一个消费者队列可以有多个消费者实例，只有其中一个消费者实例会消费。

如图所示：

技术图片

生产者：

 

package com.bunny.rabbit.work;

import com.bunny.rabbit.LianjieGongjulei;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;

/**
 * 
 * @author bunny～～我是兔子我会喵，我叫喵星兔。
 *
 */
public class Shengchanzhe {
    public static void main(String[] args) throws Exception {
        Connection c = LianjieGongjulei.getConnection();
        Channel ch = c.createChannel();
        ch.queueDeclare(LianjieGongjulei.WORK_QUEUE_NAME, false, false, false, null);
        for (int i = 0; i < 100; i++, Thread.sleep(100)) {
            String mes = "kitty" + i;
            ch.basicPublish("", LianjieGongjulei.WORK_QUEUE_NAME, null, mes.getBytes());
            System.out.println(" bunny Sent ‘" + mes + "‘");
        }
        ch.close();
        c.close();
    }
}

 

消费者1：

 

package com.bunny.rabbit.work;

import com.bunny.rabbit.LianjieGongjulei;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.QueueingConsumer;

/**
 * 
 * @author bunny～～我是兔子我会喵，我叫喵星兔。
 *
 */
public class Xiaofeizhe1 {
    public static void main(String[] args) throws Exception {
        Connection c = LianjieGongjulei.getConnection();
        Channel ch = c.createChannel();
        ch.queueDeclare(LianjieGongjulei.WORK_QUEUE_NAME, false, false, false, null);

        QueueingConsumer consumer = new QueueingConsumer(ch);
        // 监听队列，false表示手动返回完成状态，true表示自动
        ch.basicConsume(LianjieGongjulei.WORK_QUEUE_NAME, true, consumer);

        while (true) {
            QueueingConsumer.Delivery delivery = consumer.nextDelivery();
            String message = new String(delivery.getBody());
            System.out.println(" kitty Received ‘" + message + "‘");
            Thread.sleep(100);

        }

    }
}

 

消费者2：

 

package com.bunny.rabbit.work;

import java.io.IOException;

import com.bunny.rabbit.LianjieGongjulei;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.QueueingConsumer;

/**
 * 
 * @author bunny～～我是兔子我会喵，我叫喵星兔。
 *
 */
public class Xiaofeizhe2 {
    public static void main(String[] args) throws Exception {
        Connection c = LianjieGongjulei.getConnection();
        Channel ch = c.createChannel();
        ch.queueDeclare(LianjieGongjulei.WORK_QUEUE_NAME, false, false, false, null);

        //手工确认用
        ch.basicQos(1);

        // 定义队列的消费者
        QueueingConsumer consumer = new QueueingConsumer(ch);
        // 监听队列，false表示手动返回完成状态，true表示自动
        ch.basicConsume(LianjieGongjulei.WORK_QUEUE_NAME, false, consumer);

        while (true) {
            QueueingConsumer.Delivery delivery = consumer.nextDelivery();
            String message = new String(delivery.getBody());
            System.out.println(" bunny Received ‘" + message + "‘");

            Thread.sleep(100);
            // 表示手动返回完成状态,如果没有下面的一行，将只能获取到上面ch.basicQos(1)里面写的数量
            ch.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
        }

    }
}

 

 

三、通过交换机

1、发布订阅

1个生产者，多个消费者。

每一个消费者都有自己的一个队列。

生产者没有将消息直接发送到队列，而是发送到了交换机。

每个队列都要绑定到交换机。

生产者发送的消息，经过交换机，到达队列，实现一个消息被多个消费者获取的目的。

消息发送到没有队列绑定的交换机时，消息将丢失，因为，交换机没有存储消息的能力，消息只能存在在队列中。

如图：

技术图片

 

技术图片

生产者：

package com.bunny.rabbit.pubsub;

import com.bunny.rabbit.LianjieGongjulei;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
/**
 * 
 * @author bunny～～我是兔子我会喵，我叫喵星兔。
 *
 */
public class Fabuzhe {

    public static void main(String[] args) throws Exception {
        Connection c = LianjieGongjulei.getConnection();
        Channel ch = c.createChannel();
        //交换器的类型，常见的有direct,fanout,topic等，其中“fanout”是发布订阅
        ch.exchangeDeclare(LianjieGongjulei.PUBSUB_EXCHANGE_NAME, "fanout");
        String mes = "发布订阅信息";
        ch.basicPublish(LianjieGongjulei.PUBSUB_EXCHANGE_NAME, "", null, mes.getBytes());
        System.out.println(" pubsub Sent ‘" + mes + "‘");
        ch.close();
        c.close();
    
    }
 
}

消费者1：

package com.bunny.rabbit.pubsub;

import com.bunny.rabbit.LianjieGongjulei;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.QueueingConsumer;
/**
 * 
 * @author bunny～～我是兔子我会喵，我叫喵星兔。
 *
 */
public class Dingyuezhe1 {
    public static void main(String[] args) throws Exception {
        Connection c = LianjieGongjulei.getConnection();
        Channel ch = c.createChannel();
        ch.queueDeclare(LianjieGongjulei.PUBSUB_QUEUE_NAME1, false, false, false, null);

        // 绑定队列到交换机
        ch.queueBind(LianjieGongjulei.PUBSUB_QUEUE_NAME1, LianjieGongjulei.PUBSUB_EXCHANGE_NAME, "");
        
        // 监听队列，手动返回完成
        ch.basicQos(1);
        QueueingConsumer consumer = new QueueingConsumer(ch);
        // 监听队列，手动返回完成
        ch.basicConsume(LianjieGongjulei.PUBSUB_QUEUE_NAME1, false, consumer);

        while (true) {
            QueueingConsumer.Delivery delivery = consumer.nextDelivery();
            String mes = new String(delivery.getBody());
            System.out.println(" 订阅者1  Received ‘" + mes + "‘");
            Thread.sleep(100);
            // 监听队列，手动返回完成
            ch.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
        }

    }
}

消费者2：

package com.bunny.rabbit.pubsub;

import java.io.IOException;

import com.bunny.rabbit.LianjieGongjulei;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.QueueingConsumer;
/**
 * 
 * @author bunny～～我是兔子我会喵，我叫喵星兔。
 *
 */
public class Dingyuezhe2 {
    public static void main(String[] args) throws Exception {
        Connection c = LianjieGongjulei.getConnection();
        Channel ch = c.createChannel();
        ch.queueDeclare(LianjieGongjulei.PUBSUB_QUEUE_NAME2, false, false, false, null);

        // 绑定队列到交换机
        ch.queueBind(LianjieGongjulei.PUBSUB_QUEUE_NAME2, LianjieGongjulei.PUBSUB_EXCHANGE_NAME, "");

        // 监听队列，手动返回完成
        ch.basicQos(1);
        QueueingConsumer consumer = new QueueingConsumer(ch);
        // 监听队列，手动返回完成
        ch.basicConsume(LianjieGongjulei.PUBSUB_QUEUE_NAME2, false, consumer);

        while (true) {
            QueueingConsumer.Delivery delivery = consumer.nextDelivery();
            String mes = new String(delivery.getBody());
            System.out.println(" 订阅者2  Received ‘" + mes + "‘");
            Thread.sleep(100);
            // 监听队列，手动返回完成
            ch.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
        }

    }
}

 

2、路由模式

在前面基础上，改变交换机类型，添加key信息。

如图：

技术图片

技术图片

生产者：

package com.bunny.rabbit.routing;

import com.bunny.rabbit.LianjieGongjulei;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
/**
 * 
 * @author bunny～～我是兔子我会喵，我叫喵星兔。
 *
 */
public class Fabuzhe {

    public static void main(String[] args) throws Exception {
        Connection c = LianjieGongjulei.getConnection();
        Channel ch = c.createChannel();
        //交换器的类型，常见的有direct,fanout,topic等，其中“direct”是路由
        ch.exchangeDeclare(LianjieGongjulei.ROUTING_EXCHANGE_NAME, "direct");
        String mes1 = "路由直连信息1";
        String mes2 = "路由直连信息2";
        //第二个参数是消息key用于路由用
        ch.basicPublish(LianjieGongjulei.ROUTING_EXCHANGE_NAME, "meskey1", null, mes1.getBytes());
        ch.basicPublish(LianjieGongjulei.ROUTING_EXCHANGE_NAME, "meskey2", null, mes2.getBytes());
        System.out.println(" pubsub Sent ‘" + mes1 + "‘");
        System.out.println(" pubsub Sent ‘" + mes2 + "‘");
        ch.close();
        c.close();
    
    }
 
}

消费者1：

package com.bunny.rabbit.routing;

import com.bunny.rabbit.LianjieGongjulei;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.QueueingConsumer;
/**
 * 
 * @author bunny～～我是兔子我会喵，我叫喵星兔。
 *
 */
public class Dingyuezhe1 {
    public static void main(String[] args) throws Exception {
        Connection c = LianjieGongjulei.getConnection();
        Channel ch = c.createChannel();
        ch.queueDeclare(LianjieGongjulei.ROUTING_QUEUE_NAME1, false, false, false, null);

        // 绑定队列到交换机，第三个参数是key
        ch.queueBind(LianjieGongjulei.ROUTING_QUEUE_NAME1, LianjieGongjulei.ROUTING_EXCHANGE_NAME, "meskey1");
        ch.queueBind(LianjieGongjulei.ROUTING_QUEUE_NAME1, LianjieGongjulei.ROUTING_EXCHANGE_NAME, "meskey2");
        
        // 监听队列，手动返回完成
        ch.basicQos(1);
        QueueingConsumer consumer = new QueueingConsumer(ch);
        // 监听队列，手动返回完成
        ch.basicConsume(LianjieGongjulei.ROUTING_QUEUE_NAME1, false, consumer);

        while (true) {
            QueueingConsumer.Delivery delivery = consumer.nextDelivery();
            String mes = new String(delivery.getBody());
            System.out.println(" 订阅者1  Received ‘" + mes + "‘");
            Thread.sleep(100);
            // 监听队列，手动返回完成
            ch.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
        }

    }
}

消费者2：

package com.bunny.rabbit.routing;

import java.io.IOException;

import com.bunny.rabbit.LianjieGongjulei;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.QueueingConsumer;
/**
 * 
 * @author bunny～～我是兔子我会喵，我叫喵星兔。
 *
 */
public class Dingyuezhe2 {
    public static void main(String[] args) throws Exception {
        Connection c = LianjieGongjulei.getConnection();
        Channel ch = c.createChannel();
        ch.queueDeclare(LianjieGongjulei.ROUTING_QUEUE_NAME2, false, false, false, null);

        // 绑定队列到交换机
        ch.queueBind(LianjieGongjulei.ROUTING_QUEUE_NAME2, LianjieGongjulei.ROUTING_EXCHANGE_NAME, "meskey1");

        // 监听队列，手动返回完成
        ch.basicQos(1);
        QueueingConsumer consumer = new QueueingConsumer(ch);
        // 监听队列，手动返回完成
        ch.basicConsume(LianjieGongjulei.ROUTING_QUEUE_NAME2, false, consumer);

        while (true) {
            QueueingConsumer.Delivery delivery = consumer.nextDelivery();
            String mes = new String(delivery.getBody());
            System.out.println(" 订阅者2  Received ‘" + mes + "‘");
            Thread.sleep(100);
            // 监听队列，手动返回完成
            ch.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
        }

    }
}

 

3、主题模式

在路由基础上，模糊匹配。至于匹配规则，简单来说就是*代表一个单词，#代表多个单词。单词指的是字符串被“.”分割的部分。可以实际操作之后观察结果。

 如图：

技术图片

技术图片

生产者：

package com.bunny.rabbit.topic;

import com.bunny.rabbit.LianjieGongjulei;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
/**
 * 
 * @author bunny～～我是兔子我会喵，我叫喵星兔。
 *
 */
public class Fabuzhe {

    public static void main(String[] args) throws Exception {
        Connection c = LianjieGongjulei.getConnection();
        Channel ch = c.createChannel();
        //交换器的类型，常见的有direct,fanout,topic等，其中“direct”是路由
        ch.exchangeDeclare(LianjieGongjulei.TOPIC_EXCHANGE_NAME, "topic");
        String mes1 = "主题信息1";
        String mes2 = "主题信息2";
        String mes3 = "主题信息3";
        String mes4 = "主题信息4";
        //第二个参数是消息key用于路由用
        ch.basicPublish(LianjieGongjulei.TOPIC_EXCHANGE_NAME, "mes.key.bunny", null, mes1.getBytes());
        ch.basicPublish(LianjieGongjulei.TOPIC_EXCHANGE_NAME, "bunny", null, mes2.getBytes());
        ch.basicPublish(LianjieGongjulei.TOPIC_EXCHANGE_NAME, "mes", null, mes3.getBytes());
        ch.basicPublish(LianjieGongjulei.TOPIC_EXCHANGE_NAME, "key.ee", null, mes4.getBytes());
        System.out.println(" topic Sent ‘" + mes1 + "‘");
        System.out.println(" topic Sent ‘" + mes2 + "‘");
        System.out.println(" topic Sent ‘" + mes3 + "‘");
        System.out.println(" topic Sent ‘" + mes4 + "‘");
        ch.close();
        c.close();
    
    }
 
}

消费者1：

package com.bunny.rabbit.topic;

import com.bunny.rabbit.LianjieGongjulei;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.QueueingConsumer;
/**
 * 
 * @author bunny～～我是兔子我会喵，我叫喵星兔。
 *
 */
public class Dingyuezhe1 {
    public static void main(String[] args) throws Exception {
        Connection c = LianjieGongjulei.getConnection();
        Channel ch = c.createChannel();
        ch.queueDeclare(LianjieGongjulei.TOPIC_QUEUE_NAME1, false, false, false, null);

        // 绑定队列到交换机，第三个参数是key
        ch.queueBind(LianjieGongjulei.TOPIC_QUEUE_NAME1, LianjieGongjulei.TOPIC_EXCHANGE_NAME, "*");
        ch.queueBind(LianjieGongjulei.TOPIC_QUEUE_NAME1, LianjieGongjulei.TOPIC_EXCHANGE_NAME, "mes.#");
        
        // 监听队列，手动返回完成
        ch.basicQos(1);
        QueueingConsumer consumer = new QueueingConsumer(ch);
        // 监听队列，手动返回完成
        ch.basicConsume(LianjieGongjulei.TOPIC_QUEUE_NAME1, false, consumer);

        while (true) {
            QueueingConsumer.Delivery delivery = consumer.nextDelivery();
            String mes = new String(delivery.getBody());
            System.out.println(" 订阅者1  Received ‘" + mes + "‘");
            Thread.sleep(100);
            // 监听队列，手动返回完成
            ch.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
        }

    }
}

消费者2：

 

package com.bunny.rabbit.topic;

import java.io.IOException;

import com.bunny.rabbit.LianjieGongjulei;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.QueueingConsumer;
/**
 * 
 * @author bunny～～我是兔子我会喵，我叫喵星兔。
 *
 */
public class Dingyuezhe2 {
    public static void main(String[] args) throws Exception {
        Connection c = LianjieGongjulei.getConnection();
        Channel ch = c.createChannel();
        ch.queueDeclare(LianjieGongjulei.TOPIC_QUEUE_NAME2, false, false, false, null);

        // 绑定队列到交换机
        ch.queueBind(LianjieGongjulei.TOPIC_QUEUE_NAME2, LianjieGongjulei.TOPIC_EXCHANGE_NAME, "#");

        // 监听队列，手动返回完成
        ch.basicQos(1);
        QueueingConsumer consumer = new QueueingConsumer(ch);
        // 监听队列，手动返回完成
        ch.basicConsume(LianjieGongjulei.TOPIC_QUEUE_NAME2, false, consumer);

        while (true) {
            QueueingConsumer.Delivery delivery = consumer.nextDelivery();
            String mes = new String(delivery.getBody());
            System.out.println(" 订阅者2  Received ‘" + mes + "‘");
            Thread.sleep(100);
            // 监听队列，手动返回完成
            ch.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
        }

    }
}

 


四、总结


队列中数据只会被处理一次，交换机的数据处理次数和发送的队列有关。所以针对消息队列而言，交换机不是必须的。

每一个消息队列中的数据只会被消费一次。交换机把数据发到几个队列，那么数据就应该被消费几次，完全取决于队列数量。

上面的所有涉及路由的都只有一个消费者对应一个队列，其实都可以是多个消费者对应一个队列。

无论是生产者还是消费者，创建队列和交换机、队列和交换机的绑定，这些操作都是一次性的，如果第二次运行会发现已经处理了就不需要作。同样的，改变代码后并没有解除之前的设置、绑定等。不会因为有创建命令就重新创建。所以如果一个交换机开始设置成路由，那么代码修改为topic将会失败。

上面任何一步使用队列或者交换机，如果之前没创建，将会报错，这就意味着上面的代码有执行顺序。但是按照不报错的执行也不行，因为有些没效果，可以考虑先从头到尾执行一遍，无视错误。然后把运行中的终端关掉，这样环境就准备好了。然后在运行查看结果。

 

 

ps:项目地址  svn://47.105.188.20/kitty/2%E3%80%81code/rabbitclient　　用户名密码：reader/reader

 


喵星之旅-狂奔的兔子-rabbitmq的java客户端使用入门

标签：工具   添加   代码   etc   单词   快速   有一个   针对   code   

原文地址：https://www.cnblogs.com/kittybunny/p/12374168.html
踩
(0)
赞
(0)
   
举报
评论 一句话评论（0）
登录后才能评论！
分享档案
更多>
2021年07月29日 (22)
2021年07月28日 (40)
2021年07月27日 (32)
2021年07月26日 (79)
2021年07月23日 (29)
2021年07月22日 (30)
2021年07月21日 (42)
2021年07月20日 (16)
2021年07月19日 (90)
2021年07月16日 (35)
周排行
mamicode.com排行更多图片
更多
友情链接
兰亭集智   国之画   百度统计   站长统计   阿里云   chrome插件   新版天听网
关于我们 - 联系我们 - 留言反馈
© 2014 mamicode.com 版权所有  联系我们:gaon5@hotmail.com
迷上了代码！