您当前的位置:首页 > 电脑百科 > 程序开发 > 语言 > JAVA

RabbitMq七种工作模式,结合简单的java实例使用,别再说你不会

时间:2020-08-01 15:16:29  来源:  作者:

一、Maven依赖添加

 <!-- rabbitmq相关依赖 -->
        <dependency>
            <groupId>com.rabbitmq</groupId>
            <artifactId>amqp-client</artifactId>
            <version>3.0.4</version>
        </dependency>

二、七种工作模式的JAVA实例

1、简单模式

最简单的一个消费者和一个生产者模式,生产者生成消息,消费者监听消息,若是消费者监听到它所需要的消息,就会消费该消息,这种消息是次性的,被消费了就没有了。

RabbitMq七种工作模式,结合简单的java实例使用,别再说你不会

 

1.1.1、EasyRecv.java

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

public class EasyRecv {
    //队列名称
    private final static String QUEUE_NAME = "hello world";
    public static void main(String[] argv) throws java.io.IOException,java.lang.InterruptedException {
        //打开连接和创建频道,与发送端一样
        ConnectionFactory factory = new ConnectionFactory();
        //设置RabbitMQ所在主机ip或者主机名
        factory.setHost("127.0.0.1");
        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();
        //声明队列,主要为了防止消息接收者先运行此程序,队列还不存在时创建队列。
               /**
         * 队列名
         * 是否持久化
         *  是否排外  即只允许该channel访问该队列   一般等于true的话用于一个队列只能有一个消费者来消费的场景
         *  是否自动删除  消费完删除
         *  其他属性
         *
         */
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        System.out.println("Waiting for messages. To exit press CTRL+C");

        //创建队列消费者
        QueueingConsumer consumer = new QueueingConsumer(channel);
        //指定消费队列
         /**
         * 队列名
         * 其他属性  路由
         * 消息body
         */
        channel.basicConsume(QUEUE_NAME, true, consumer);
        while (true)
        {
            //nextDelivery是一个阻塞方法(内部实现其实是阻塞队列的take方法)
            QueueingConsumer.Delivery delivery = consumer.nextDelivery();
            String message = new String(delivery.getBody());
            System.out.println("Received '" + message + "'");
        }

    }
}

1.1.2、EasySend.java

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

import java.util.Scanner;

public class EasySend {

    //队列名称
    private final static String QUEUE_NAME = "hello world";

    public static void main(String[] argv) throws java.io.IOException
    {
        /**
         * 创建连接连接到MabbitMQ
         */
        ConnectionFactory factory = new ConnectionFactory();
        //设置MabbitMQ所在主机ip或者主机名
        factory.setHost("127.0.0.1");


        while (true){
            //创建一个连接
            Connection connection = factory.newConnection();
            //创建一个频道
            Channel channel = connection.createChannel();
            //指定一个队列
            channel.queueDeclare(QUEUE_NAME, false, false, false, null);
            //发送的消息
            Scanner scanner = new Scanner(System.in);
            String ms = scanner.nextLine();
            //String message = "hello world!";
            //往队列中发出一条消息
            channel.basicPublish("", QUEUE_NAME, null, ms.getBytes());
            System.out.println("Sent '" + ms + "'");
            //关闭频道和连接
            channel.close();
            connection.close();
        }
    }

以上两个已经可以进行通信了,下面同样是简单的实例,但是我们可以看到在代码层面上,连接的代码都是一样的,所以我们可以创建一个连接的工具类。

1.2.1、RabbitmqConnectionUtil .java

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

import java.io.IOException;


public  class RabbitmqConnectionUtil {

    public static Connection getConnection() throws IOException {
        //连接工厂
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        //连接5672端口  注意15672为工具界面端口  25672为集群端口
        factory.setPort(5672);
        //factory.setVirtualHost("/xxxxx");
       // factory.setUsername("xxxxxx");
       // factory.setPassword("123456");
        //获取连接
        Connection connection = factory.newConnection();
        return connection;
    }
}

1.2.2、UtilSend.java

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import top.san.RabbitMq.util.RabbitmqConnectionUtil;

import java.io.IOException;

public class UtilSend {
    private final static String QUEUE_NAME = "UtilConn";
    public static void main(String[] args) throws IOException {

        Connection connection = RabbitmqConnectionUtil.getConnection();
        //创建通道
        Channel channel = connection.createChannel();
        //声明队列
 
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);

        //消息内容
  
        String message = "这里是lbw广场";
        channel.basicPublish("", QUEUE_NAME,null,message.getBytes());
        System.out.println("[x]Sent '"+message + "'");
        //最后关闭通关和连接
        channel.close();
        connection.close();
    }
}

1.2.3、UtilRecv.java

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.QueueingConsumer;
import top.san.RabbitMq.util.RabbitmqConnectionUtil;

import java.io.IOException;

public class UtilRecv {
    private final static String QUEUE_NAME = "UtilConn";
    public static void main(String[] args) throws IOException, InterruptedException {

        Connection connection = null;
        connection = RabbitmqConnectionUtil.getConnection();
        //创建通道
        Channel channel = connection.createChannel();
        //声明队列
       
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);

        QueueingConsumer queueingConsumer = new QueueingConsumer(channel);

        channel.basicConsume(QUEUE_NAME,true,queueingConsumer);

        while(true){
            //该方法会阻塞
            QueueingConsumer.Delivery delivery = queueingConsumer.nextDelivery();
            String message = new String(delivery.getBody());
            System.out.println("[x] Received '"+message+"'");
        }

    }
}

2、工作队列

工作队列也就是简单模式的强化版,一个队列是可以多个生产者,也可以有多个消费者来竞争消费消息,但是我们仍需保证队列的幂等性,队列存在就不能再创建同名队列。

RabbitMq七种工作模式,结合简单的java实例使用,别再说你不会

 

下面的每个进程都控制其主线程休眠,让我们可以更好的看到结果。

2.1.1、Sender1.java

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import top.san.RabbitMq.util.RabbitmqConnectionUtil;

import java.io.IOException;

public class Sender1 {
    private final  static String QUEUE_NAME = "queue_work";

    public static void main(String[] args) throws IOException, InterruptedException {
        Connection connection = RabbitmqConnectionUtil.getConnection();
        Channel channel = connection.createChannel();

        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        for(int i = 0; i < 100; i++){
            String message = "lbw" + i;
            channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
            System.out.println("[x] Sent '"+message + "'");
            Thread.sleep(i*10);
        }

        channel.close();
        connection.close();
    }
}

2.1.2、Sender2.java

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import top.san.RabbitMq.util.RabbitmqConnectionUtil;

import java.io.IOException;

public class Sender2 {
    private final  static String QUEUE_NAME = "queue_work";

    public static void main(String[] args) throws IOException, InterruptedException {
        Connection connection = RabbitmqConnectionUtil.getConnection();
        Channel channel = connection.createChannel();

        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        for(int i = 0; i < 100; i++){
            String message = "nb" + i;
            channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
            System.out.println("[x] Sent '"+message + "'");
            Thread.sleep(i*10);
        }
        channel.close();
        connection.close();
    }
}

2.1.3、Receiver1.java

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.QueueingConsumer;

import top.san.RabbitMq.util.RabbitmqConnectionUtil;

import java.io.IOException;

/**
 * Created by san
 */
public class Receiver1 {
    private final static  String QUEUE_NAME = "queue_work";

    public static void main(String[] args) throws IOException, InterruptedException {
        Connection connection = RabbitmqConnectionUtil.getConnection();
        Channel channel = connection.createChannel();

        channel.queueDeclare(QUEUE_NAME, false,false, false,null);
        //同一时刻服务器只会发送一条消息给消费者
        channel.basicQos(1);

        QueueingConsumer consumer = new QueueingConsumer(channel);
        //关于手工确认 待之后有时间研究下
        channel.basicConsume(QUEUE_NAME, false, consumer);

        while(true){
            QueueingConsumer.Delivery delivery = consumer.nextDelivery();
            String message = new String(delivery.getBody());
            System.out.println("[x] Received1 '"+message+"'");
            Thread.sleep(10);
            //返回确认状态
            channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
        }

    }
}

2.1.4、Receiver2.java

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.QueueingConsumer;

import top.san.RabbitMq.util.RabbitmqConnectionUtil;

import java.io.IOException;

/**
 * Created by san
 */
public class Receiver2 {
    private final static  String QUEUE_NAME = "queue_work";

    public static void main(String[] args) throws IOException, InterruptedException {

        Connection connection = RabbitmqConnectionUtil.getConnection();
        Channel channel = connection.createChannel();

        channel.queueDeclare(QUEUE_NAME, false,false, false,null);
        //同一时刻服务器只会发送一条消息给消费者
        channel.basicQos(1);

        QueueingConsumer consumer = new QueueingConsumer(channel);
        
        
        channel.basicConsume(QUEUE_NAME, false, consumer);

        while(true){
            QueueingConsumer.Delivery delivery = consumer.nextDelivery();
            String message = new String(delivery.getBody());
            System.out.println("[x] Received2 '"+message+"'");
            Thread.sleep(1000);
            //返回确认状态
            channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
        }

    }
}

2.1.5、结果

上面的四个程序都运行起来,结果可以看到如下,依据结果分析,可知,同一个消息队列,是可以有多个生产者和消费者的。

RabbitMq七种工作模式,结合简单的java实例使用,别再说你不会

 


RabbitMq七种工作模式,结合简单的java实例使用,别再说你不会

 


RabbitMq七种工作模式,结合简单的java实例使用,别再说你不会

 


RabbitMq七种工作模式,结合简单的java实例使用,别再说你不会

 

3、发布/订阅(fanout)

RabbitMq七种工作模式,结合简单的java实例使用,别再说你不会

 

3.1.1、Sender.java

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import top.san.RabbitMq.util.RabbitmqConnectionUtil;

public class Sender {
    private final static String EXCHANGE_NAME = "test_exchange_fanout";

    public static void main(String[] args)
    {
        try
        {
            //获取连接
            Connection connection = RabbitmqConnectionUtil.getConnection();
            //从连接中获取一个通道
            Channel channel = connection.createChannel();
            //声明交换机(分发:发布/订阅模式)
            channel.exchangeDeclare(EXCHANGE_NAME, "fanout");
            //发送消息
            for (int i = 0; i < 5; i++)
            {
                String message = "卢本伟广场" + i;
                System.out.println("[send]:" + message);
                //发送消息
                channel.basicPublish(EXCHANGE_NAME, "", null, message.getBytes("utf-8"));
                Thread.sleep(5 * i);
            }
            channel.close();
            connection.close();
        }
        catch (Exception e)
        {
            e.printStackTrace();
        }
    }

}

3.1.2、Receiver1.java

import com.rabbitmq.client.*;
import top.san.RabbitMq.util.RabbitmqConnectionUtil;

import java.io.IOException;

public class Receiver1 {
    //交换机名称
    private final static String EXCHANGE_NAME = "test_exchange_fanout";

    //队列名称
    private static final String QUEUE_NAME    = "test_queue_email";

    public static void main(String[] args)
    {
        try
        {
            //获取连接
            Connection connection = RabbitmqConnectionUtil.getConnection();
            //从连接中获取一个通道
            final Channel channel = connection.createChannel();
            //声明交换机(分发:发布/订阅模式)
            channel.exchangeDeclare(EXCHANGE_NAME, "fanout");
            //声明队列
            channel.queueDeclare(QUEUE_NAME, false, false, false, null);
            //将队列绑定到交换机
            channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "");
            //保证一次只分发一个
            int prefetchCount = 1;
            channel.basicQos(prefetchCount);
            //定义消费者
            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("[email] Receive message:" + message);
                    try
                    {
                        //消费者休息2s处理业务
                        Thread.sleep(1000);
                    }
                    catch (InterruptedException e)
                    {
                        e.printStackTrace();
                    }
                    finally
                    {
                        System.out.println("[1] done");
                        //手动应答
                        channel.basicAck(envelope.getDeliveryTag(), false);
                    }
                }
            };
            //设置手动应答
            boolean autoAck = false;
            //监听队列
            channel.basicConsume(QUEUE_NAME, autoAck, consumer);
        }
        catch (IOException e)
        {
            e.printStackTrace();
        }
    }
}

3.1.3、Receiver2.java

import com.rabbitmq.client.*;
import top.san.RabbitMq.util.RabbitmqConnectionUtil;

import java.io.IOException;

public class Receiver2 {

    //交换机名称
    private final static String EXCHANGE_NAME = "test_exchange_fanout";

    //队列名称
    private static final String QUEUE_NAME    = "test_queue_phone";

    public static void main(String[] args)
    {
        try
        {

            //获取连接
            Connection connection = RabbitmqConnectionUtil.getConnection();
            //从连接中获取一个通道
            final Channel channel = connection.createChannel();
            //声明交换机(分发:发布/订阅模式)
            channel.exchangeDeclare(EXCHANGE_NAME, "fanout");
            //声明队列
            channel.queueDeclare(QUEUE_NAME, false, false, false, null);
            //将队列绑定到交换机
            channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "");
            //保证一次只分发一个
            int prefetchCount = 1;
            channel.basicQos(prefetchCount);
            //定义消费者
            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("[phone] Receive message:" + message);
                    try
                    {
                        //消费者休息1s处理业务
                        Thread.sleep(1000);
                    }
                    catch (InterruptedException e)
                    {
                        e.printStackTrace();
                    }
                    finally
                    {
                        System.out.println("[2] done");
                        //手动应答
                        channel.basicAck(envelope.getDeliveryTag(), false);
                    }
                }
            };
            //设置手动应答
            boolean autoAck = false;
            //监听队列
            channel.basicConsume(QUEUE_NAME, autoAck, consumer);
        }
        catch (IOException e)
        {
            e.printStackTrace();
        }
    }
}

3.1.4、结果

从程序运行结果和RabbitMq的后台看出,这样的消息属于广播型,两个不同名的队列的都能收到该消息,只需它们都将自己绑定到同一个交换机,而且,该消息是持久的,只要交换机还在,消费者啥时候上线都能消费它所绑定的交换机,而且只会一个消费者只会消费一

RabbitMq七种工作模式,结合简单的java实例使用,别再说你不会

 


RabbitMq七种工作模式,结合简单的java实例使用,别再说你不会

 


RabbitMq七种工作模式,结合简单的java实例使用,别再说你不会

 


RabbitMq七种工作模式,结合简单的java实例使用,别再说你不会

 

4、路由(direct)

  1. 在前面的示例中,我们已经在创建绑定。您可能会想起类似的代码:
channel.queueBind(queueName,EXCHANGE_NAME,“”);

绑定是交换和队列之间的关系。可以简单地理解为:队列对来自此交换的消息感兴趣。

  1. 绑定可以采用额外的routingKey参数。为了避免与basic_publish参数混淆,我们将其称为 绑定键。这是我们可以创建带有键的绑定的方法:
channel.queueBind(queueName,EXCHANGE_NAME,“ black”);
  • 直接绑定(密钥直接绑定到单个队列)
RabbitMq七种工作模式,结合简单的java实例使用,别再说你不会

 

  • 多重绑定(相同的绑定密钥绑定多个队列)
RabbitMq七种工作模式,结合简单的java实例使用,别再说你不会

 

  • 不同密钥绑定不同的队列,可以发挥出不同日志级别发送到不同的队列的效果。
RabbitMq七种工作模式,结合简单的java实例使用,别再说你不会

 

4.1.1、Sender

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import top.san.RabbitMq.util.RabbitmqConnectionUtil;

import java.io.IOException;


public class Sender {
    private final static String EXCHANGE_NAME = "exchange_direct";
    private final static String EXCHANGE_TYPE = "direct";

    public static void main(String[] args) throws IOException {
        Connection connection = RabbitmqConnectionUtil.getConnection();
        Channel channel = connection.createChannel();

        channel.exchangeDeclare(EXCHANGE_NAME,EXCHANGE_TYPE);

        String message = "那一定是蓝色";
        channel.basicPublish(EXCHANGE_NAME,"key2", null, message.getBytes());
        System.out.println("[x] Sent '"+message+"'");

        channel.close();
        connection.close();
    }
}

4.1.2、Receiver1.java

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.QueueingConsumer;
import top.san.RabbitMq.util.RabbitmqConnectionUtil;

import java.io.IOException;

/**
 * Created by san
 */
public class Receiver1 {
    private final  static  String QUEUE_NAME = "queue_routing";
    private final static String EXCHANGE_NAME = "exchange_direct";

    public static void main(String[] args) throws IOException, InterruptedException {
        // 获取到连接以及mq通道
        Connection connection = RabbitmqConnectionUtil.getConnection();
        Channel channel = connection.createChannel();

        channel.queueDeclare(QUEUE_NAME, false,false,false,null);
        channel.queueBind(QUEUE_NAME,EXCHANGE_NAME,"key");
        channel.queueBind(QUEUE_NAME,EXCHANGE_NAME,"key2");

        channel.basicQos(1);

        QueueingConsumer consumer = new QueueingConsumer(channel);
        channel.basicConsume(QUEUE_NAME, false, consumer);

        while(true){
            QueueingConsumer.Delivery delivery = consumer.nextDelivery();
            String message = new String(delivery.getBody());
            System.out.println("[x] Received1 "+message);
            Thread.sleep(10);

            channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
        }


    }
}

4.1.3、Receiver2.java

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.QueueingConsumer;
import top.san.RabbitMq.util.RabbitmqConnectionUtil;
import java.io.IOException;

/**
 * Created by san
 */
public class Receiver2 {
    private final  static  String QUEUE_NAME = "queue_routing2";
    private final static String EXCHANGE_NAME = "exchange_direct";

    public static void main(String[] args) throws IOException, InterruptedException {
        // 获取到连接以及mq通道
        Connection connection = RabbitmqConnectionUtil.getConnection();
        Channel channel = connection.createChannel();

        channel.queueDeclare(QUEUE_NAME, false,false,false,null);
        channel.queueBind(QUEUE_NAME,EXCHANGE_NAME,"key2");

        channel.basicQos(1);

        QueueingConsumer consumer = new QueueingConsumer(channel);
        channel.basicConsume(QUEUE_NAME, false, consumer);

        while(true){
            QueueingConsumer.Delivery delivery = consumer.nextDelivery();
            String message = new String(delivery.getBody());
            System.out.println("[x] Received2 "+message);
            Thread.sleep(10);

            channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
        }
    }
}

4.1.4、结果-总结

有一点要注意是:在direct下,必须是Exchange(交换机)已经存在,消费端的队列才能绑定到Exchange,否则会报错。也就说上面的程序第一次运行时,需先启Sender,才能成功启动Reciver。

RabbitMq七种工作模式,结合简单的java实例使用,别再说你不会

 


RabbitMq七种工作模式,结合简单的java实例使用,别再说你不会

 


RabbitMq七种工作模式,结合简单的java实例使用,别再说你不会

 


RabbitMq七种工作模式,结合简单的java实例使用,别再说你不会

 


RabbitMq七种工作模式,结合简单的java实例使用,别再说你不会

 

5、话题(topic)

话题也是一个持久的消息,只要交换机还在,每个上线的消费者都可以消费一次自己感兴趣的topic。

  • *(星号)可以代替一个单词。
  • #(哈希)可以替代零个或多个单词。
RabbitMq七种工作模式,结合简单的java实例使用,别再说你不会

 

5.1.1、Sender.java

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import top.san.RabbitMq.util.RabbitmqConnectionUtil;

import java.io.IOException;

public class Sender {
    private final static String EXCHANGE_NAME = "exchange_topic";
    private final static String EXCHANGE_TYPE = "topic";

    public static void main(String[] args) throws IOException {
        Connection connection = RabbitmqConnectionUtil.getConnection();
        Channel channel = connection.createChannel();

        channel.exchangeDeclare(EXCHANGE_NAME, EXCHANGE_TYPE);

        //消息内容
        String message = "这里是卢本伟广场";
        //第二个参数是topic匹配值
        channel.basicPublish(EXCHANGE_NAME,"lbw.nb",null,message.getBytes());
        System.out.println("[x] Sent '"+message+"'");

        //关通道 关连接
        channel.close();
        connection.close();
    }
}

5.1.2、Receiver1.java

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.QueueingConsumer;
import top.san.RabbitMq.util.RabbitmqConnectionUtil;

import java.io.IOException;

public class Receiver1 {
    private final static String QUEUE_NAME = "queue_topic";
    private final static String EXCHANGE_NAME = "exchange_topic";
    private final static String EXCHANGE_TYPE = "topic";

    public static void main(String[] args) throws IOException, InterruptedException {
        Connection connection = RabbitmqConnectionUtil.getConnection();
        Channel channel = connection.createChannel();

        channel.queueDeclare(QUEUE_NAME, false, false,false, null);
        //第二参数就是去匹配我兴趣的topic
        channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "lbw.nb.*");

        channel.basicQos(1);

        QueueingConsumer consumer = new QueueingConsumer(channel);
        channel.basicConsume(QUEUE_NAME, false, consumer);

        while(true){
            QueueingConsumer.Delivery delivery = consumer.nextDelivery();
            String message = new String(delivery.getBody());
            System.out.println("[x] Received1 '"+message + "'");
            Thread.sleep(10);
            channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);

        }
    }
}

5.1.3、Receiver2.java

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.QueueingConsumer;
import top.san.RabbitMq.util.RabbitmqConnectionUtil;

import java.io.IOException;

public class Receiver2 {

    private final static String QUEUE_NAME = "queue_topic2";
    private final static String EXCHANGE_NAME = "exchange_topic";
    private final static String EXCHANGE_TYPE = "topic";

    public static void main(String[] args) throws IOException, InterruptedException {
        Connection connection = RabbitmqConnectionUtil.getConnection();
        Channel channel = connection.createChannel();

        channel.queueDeclare(QUEUE_NAME, false, false,false, null);

        //第二参数就是去匹配我兴趣的topic
        channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "lbw.#");

        channel.basicQos(1);

        QueueingConsumer consumer = new QueueingConsumer(channel);
        channel.basicConsume(QUEUE_NAME, false, consumer);

        while(true){
            QueueingConsumer.Delivery delivery = consumer.nextDelivery();
            String message = new String(delivery.getBody());
            System.out.println("[x] Received2 '"+message + "'");
            Thread.sleep(10);

            channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
        }
    }
}

5.1.4、结果-分析

话题的特色就是队列可以获取自己感兴趣的话题消息,可以通过通配符*或#来表示匹配所有的感兴趣的字符串。

RabbitMq七种工作模式,结合简单的java实例使用,别再说你不会

 


RabbitMq七种工作模式,结合简单的java实例使用,别再说你不会

 


RabbitMq七种工作模式,结合简单的java实例使用,别再说你不会

 


RabbitMq七种工作模式,结合简单的java实例使用,别再说你不会

 


RabbitMq七种工作模式,结合简单的java实例使用,别再说你不会

 

6、RPC(远程过程调用)

给张图自己体会吧(官网没给示例代码,我也就不写了),就是通过两个交换机实现一个可回调的过程吧。

RabbitMq七种工作模式,结合简单的java实例使用,别再说你不会

 

三、RabbitMq的交换机

RabbitMq是有一个交换机的概念的, 消息(Message)由Client发送,RabbitMQ接收到消息之后通过交换机转发到对应的队列上面。Worker会从队列中获取未被读取的数据处理。这样就可以实现消息的发送者无需知道消息使用者的存在,反之亦然。Direct exchange:直连(路由)交换机,转发消息到routigKey指定的队列

Fanout exchange:扇形交换机,转发消息到所有绑定队列(相当于广播)

Topic exchange:主题交换机,按规则转发消息(很灵活)

Headers exchange:首部交换机

RabbitMq七种工作模式,结合简单的java实例使用,别再说你不会

 

前面的简单类型我们都是忽略了交换机的参数的,如该方法:channel.basicPublish("", QUEUE_NAME, null, message.getBytes());就是这个方法的第一个参数,置空说明使用了默认的交换机。
有几种交换类型可用:direct,topic,headers 和fanout。

作者:小小卡尔

原文链接:https://blog.csdn.net/weixin_44185736/article/details/106574637



Tags:RabbitMq   点击:()  评论:()
声明:本站部分内容及图片来自互联网,转载是出于传递更多信息之目的,内容观点仅代表作者本人,如有任何标注错误或版权侵犯请与我们联系(Email:2595517585@qq.com),我们将及时更正、删除,谢谢。
▌相关推荐
RabbitMQ 介绍RabbitMQ 是一个由erlang语言编写的、开源的、在AMQP基础上完整的、可复用的企业消息系统。支持多种语言,包括java、Python、ruby、PHP、C/C++等。1.1.AMQP模型...【详细内容】
2021-11-17  Tags: RabbitMq  点击:(16)  评论:(0)  加入收藏
下载Erlang和RabbitMQ官网下载地址Erlang下载地址: http://www.erlang.org/downloadsRabbitMQ下载地址: http://www.rabbitmq.com/download.html版本:( Erlang23+RabbitMQ3.8.4...【详细内容】
2021-08-31  Tags: RabbitMq  点击:(49)  评论:(0)  加入收藏
环境:Spring Boot2.3.10 + RabbitMQ 3.8.12 + Erlang 23.2.51.1 RabbitMQ介绍RabbitMQ是实现了高级消息队列协议(AMQP)的开源消息代理软件(亦称面向消息的中间件)。RabbitMQ服务...【详细内容】
2021-04-22  Tags: RabbitMq  点击:(337)  评论:(0)  加入收藏
RabbitMQ环境搭建erlang和RabbitMQ版本对应关系:https://www.rabbitmq.com/which-erlang.htmlerlang环境安装yum方式安装 yum源配置[root@iyeed RabbitMQ]# curl -s https://...【详细内容】
2021-04-14  Tags: RabbitMq  点击:(281)  评论:(0)  加入收藏
Direct 模式# 所有发送到 Direct Exchange 的消息被转发到 RouteKey 中指定的 Queue。 Direct 模式可以使用 RabbitMQ 自带的 Exchange: default Exchange,所以不需要将 Exch...【详细内容】
2021-04-13  Tags: RabbitMq  点击:(220)  评论:(0)  加入收藏
一、关于 RabbitMQ说到 RabbitMQ,相信大家都不会陌生,微服务开发中必不可少的中间件。 在上篇关于消息队列的文章中,我们了解到 RabbitMQ 本质其实是用 Erlang 开发的 AMQP(Adva...【详细内容】
2021-03-11  Tags: RabbitMq  点击:(229)  评论:(0)  加入收藏
说明:想要理解RabbitMQ,需要先理解MQ是什么?能做什么?然后根据基础知识去理解RabbitMQ是什么、提供了什么功能。一、MQ的简单理解1. 什么是MQ? 消息队列(Message Queue),是基础数据...【详细内容】
2021-02-07  Tags: RabbitMq  点击:(155)  评论:(0)  加入收藏
1、 查找Docker容器中的RabbitMQ镜像docker ps -a[root@linux ~]# docker ps -aCONTAINER ID IMAGE COMMAND CREATED...【详细内容】
2020-11-27  Tags: RabbitMq  点击:(221)  评论:(0)  加入收藏
一、Maven依赖添加 <!-- rabbitmq相关依赖 --> <dependency> <groupId>com.rabbitmq</groupId> <artifactId>amqp-client</artifactId>...【详细内容】
2020-08-01  Tags: RabbitMq  点击:(53)  评论:(0)  加入收藏
一、简单的发送与接收消息 HelloWorld1. 发送消息发送消息首先要获取与rabbitmq-server的连接,然后从渠道(chann)中指定的queue发送消息 , 不能定义两个queue名字相同,但属性...【详细内容】
2020-04-03  Tags: RabbitMq  点击:(61)  评论:(0)  加入收藏
▌简易百科推荐
一、Redis使用过程中一些小的注意点1、不要把Redis当成数据库来使用二、Arrays.asList常见失误需求:把数组转成list集合去处理。方法:Arrays.asList 或者 Java8的stream流式处...【详细内容】
2021-12-27  CF07    Tags:Java   点击:(3)  评论:(0)  加入收藏
文章目录 如何理解面向对象编程? JDK 和 JRE 有什么区别? 如何理解Java中封装,继承、多态特性? 如何理解Java中的字节码对象? 你是如何理解Java中的泛型的? 说说泛型应用...【详细内容】
2021-12-24  Java架构师之路    Tags:JAVA   点击:(5)  评论:(0)  加入收藏
大家好!我是老码农,一个喜欢技术、爱分享的同学,从今天开始和大家持续分享JVM调优方面的经验。JVM调优是个大话题,涉及的知识点很庞大 Java内存模型 垃圾回收机制 各种工具使用 ...【详细内容】
2021-12-23  小码匠和老码农    Tags:JVM调优   点击:(11)  评论:(0)  加入收藏
前言JDBC访问Postgresql的jsonb类型字段当然可以使用Postgresql jdbc驱动中提供的PGobject,但是这样在需要兼容多种数据库的系统开发中显得不那么通用,需要特殊处理。本文介绍...【详细内容】
2021-12-23  dingle    Tags:JDBC   点击:(12)  评论:(0)  加入收藏
Java与Lua相互调用案例比较少,因此项目使用需要做详细的性能测试,本内容只做粗略测试。目前已完成初版Lua-Java调用框架开发,后期有时间准备把框架进行抽象,并开源出来,感兴趣的...【详细内容】
2021-12-23  JAVA小白    Tags:Java   点击:(10)  评论:(0)  加入收藏
Java从版本5开始,在 java.util.concurrent.locks包内给我们提供了除了synchronized关键字以外的几个新的锁功能的实现,ReentrantLock就是其中的一个。但是这并不意味着我们可...【详细内容】
2021-12-17  小西学JAVA    Tags:JAVA并发   点击:(10)  评论:(0)  加入收藏
一、概述final是Java关键字中最常见之一,表示“最终的,不可更改”之意,在Java中也正是这个意思。有final修饰的内容,就会变得与众不同,它们会变成终极存在,其内容成为固定的存在。...【详细内容】
2021-12-15  唯一浩哥    Tags:Java基础   点击:(14)  评论:(0)  加入收藏
1、问题描述关于java中的日志管理logback,去年写过关于logback介绍的文章,这次项目中又优化了下,记录下,希望能帮到需要的朋友。2、解决方案这次其实是碰到了一个问题,一般的情况...【详细内容】
2021-12-15  软件老王    Tags:logback   点击:(17)  评论:(0)  加入收藏
本篇文章我们以AtomicInteger为例子,主要讲解下CAS(Compare And Swap)功能是如何在AtomicInteger中使用的,以及提供CAS功能的Unsafe对象。我们先从一个例子开始吧。假设现在我们...【详细内容】
2021-12-14  小西学JAVA    Tags:JAVA   点击:(21)  评论:(0)  加入收藏
一、概述观察者模式,又可以称之为发布-订阅模式,观察者,顾名思义,就是一个监听者,类似监听器的存在,一旦被观察/监听的目标发生的情况,就会被监听者发现,这么想来目标发生情况到观察...【详细内容】
2021-12-13  唯一浩哥    Tags:Java   点击:(16)  评论:(0)  加入收藏
最新更新
栏目热门
栏目头条