当前位置： 首页 > news >正文

苏州浒关网站建设/广告加盟

news 2025/7/1 0:29:00
苏州浒关网站建设,广告加盟,内贸在什么网站做,海湾网站建设目录 一、交换机类型 二、简单模式 1、介绍 2、代码实现 三、Work Queues工作队列模式 1、介绍 2、代码实现 四、Pub/Sub订阅模式 1、介绍 2、代码实现 五、Routing路由模式 1、介绍 2、代码实现 六、Topics通配符模式 1、介绍 2、代码实现 一、交换机类型 在…

目录

一、交换机类型

二、简单模式

1、介绍

2、代码实现 

三、Work Queues工作队列模式

1、介绍

2、代码实现

四、Pub/Sub订阅模式

1、介绍

2、代码实现

五、Routing路由模式

1、介绍

2、代码实现

六、Topics通配符模式

1、介绍

2、代码实现

一、交换机类型

在之前的文章里我们了解了什么是交换机，当有生产者把消息发给MQ服务器时，消息会通过交换机分发给不同的消息队列里，那么交换机是如何分发消息的，我们就需要了解交换机的几种类型

类型说明
fanout广播类型，交换机会把收到的消息分发给每一与他绑定的队列
direct定向类型，交换机会根据路由将消息分发给与它绑定的指定的匹配队列
topic通配符类型，交换机会根据通配符匹配后分发给与他绑定的指定的匹配队列

二、简单模式

1、介绍

RabbitMQ的简单模式是一对一即，一个生产者生产消息后不经交换机直接给指定的队列供消费者消费

2、代码实现 

他的不需要交换机进行分发，所以我们代码实现整体过程为

生产者：建立连接-》创建队列-》构造发送消息-》关闭连接

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;import java.io.IOException;
import java.util.concurrent.TimeoutException;public class Producer {private static final String QUEUE = "easyQueue";public static void main(String[] args) throws IOException, TimeoutException {// 1.创建连接工厂 配置连接工厂ConnectionFactory factory = new ConnectionFactory();factory.setHost("127.0.0.1");factory.setPort(5672);factory.setVirtualHost("/DemoVirtualHost");factory.setUsername("guest");factory.setPassword("guest");// 2.建立连接 获取连接通道Connection connection = factory.newConnection();Channel channel = connection.createChannel();// 3.创建队列channel.queueDeclare(QUEUE,true,false,false,null);// 4.发送消息String body = "hello mq";channel.basicPublish("",QUEUE,null,body.getBytes());// 5.关闭连接channel.close();connection.close();}
}

运行代码即可发送消息到服务器 

消费者：建立连接监听队列获取消息

import com.rabbitmq.client.*;import java.io.IOException;
import java.util.concurrent.TimeoutException;public class Consumer {private static final String QUEUE = "easyQueue";public static void main(String[] args) throws IOException, TimeoutException {// 1.创建连接工厂 配置连接工厂ConnectionFactory factory = new ConnectionFactory();factory.setHost("127.0.0.1");factory.setPort(5672);factory.setVirtualHost("/DemoVirtualHost");factory.setUsername("guest");factory.setPassword("guest");// 2.建立连接Connection connection = factory.newConnection();Channel channel = connection.createChannel();// 3.监听队列获取消息com.rabbitmq.client.Consumer consumer = new DefaultConsumer(channel) {@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {System.out.println("消费者消费消息:" + new String(body));}};channel.basicConsume(QUEUE,true,consumer);}
}

 运行代码

完成消费，这就是RabbitMQ工作模式的简单模式，他不需要交换机来去对消息进行分发 

三、Work Queues工作队列模式

1、介绍

工作队列模式相比简单模式，他的处理任务速度在一定情况下会更快，因为他相比简单模式单一消费者而言它增加了消费者个数

2、代码实现

它同样也不需要交换机对收到的消息进行分发，代码实现与简单模式基本相同，唯一不同点是他相比简单模式多了一个消费者

生产者

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;import java.io.IOException;
import java.util.concurrent.TimeoutException;public class Producer {private static final String QUEUE = "workQueue";public static void main(String[] args) throws IOException, TimeoutException {// 1.创建连接工厂 配置连接工厂ConnectionFactory factory = new ConnectionFactory();factory.setHost("127.0.0.1");factory.setPort(5672);factory.setVirtualHost("/DemoVirtualHost");factory.setUsername("guest");factory.setPassword("guest");// 2.建立连接 获取连接通道Connection connection = factory.newConnection();Channel channel = connection.createChannel();// 3.创建队列channel.queueDeclare(QUEUE,true,false,false,null);// 4.发送消息String body = "hello mq";channel.basicPublish("",QUEUE,null,body.getBytes());// 5.关闭连接channel.close();connection.close();}
}

消费者1

import com.rabbitmq.client.*;import java.io.IOException;
import java.util.concurrent.TimeoutException;public class Consumer {private static final String QUEUE = "workQueue";public static void main(String[] args) throws IOException, TimeoutException {// 1.创建连接工厂 配置连接工厂ConnectionFactory factory = new ConnectionFactory();factory.setHost("127.0.0.1");factory.setPort(5672);factory.setVirtualHost("/DemoVirtualHost");factory.setUsername("guest");factory.setPassword("guest");// 2.建立连接Connection connection = factory.newConnection();Channel channel = connection.createChannel();// 3.监听队列获取消息com.rabbitmq.client.Consumer consumer = new DefaultConsumer(channel) {@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {System.out.println("消费者消费消息:" + new String(body));}};channel.basicConsume(QUEUE,true,consumer);}
}

 消费者2：

import com.rabbitmq.client.*;import java.io.IOException;
import java.util.concurrent.TimeoutException;public class Consumer {private static final String QUEUE = "workQueue";public static void main(String[] args) throws IOException, TimeoutException {// 1.创建连接工厂 配置连接工厂ConnectionFactory factory = new ConnectionFactory();factory.setHost("127.0.0.1");factory.setPort(5672);factory.setVirtualHost("/DemoVirtualHost");factory.setUsername("guest");factory.setPassword("guest");// 2.建立连接Connection connection = factory.newConnection();Channel channel = connection.createChannel();// 3.监听队列获取消息com.rabbitmq.client.Consumer consumer = new DefaultConsumer(channel) {@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {System.out.println("消费者消费消息:" + new String(body));}};channel.basicConsume(QUEUE,true,consumer);}
}

四、Pub/Sub订阅模式

1、介绍

他相比之前两种模式引入了交换机，当MQ服务器收到消息后，交换机会把收到的消息分发给每一个绑定了该交换机的队列给不同的消费者进行消费，之前的文章里我们是通过控制台进行队列与交换机的绑定，这里我们会使用Java代码来进行绑定

 比如当用户下单后我们需要去给用户通过邮箱以及短信发送回购优惠券信息时，我们就可以通过该模式，分别创建邮箱队列以及短信队列，然后生产者将用户信息发给MQ服务器经过交换机进入邮箱与短信队列让不同的消费者去消费通知。

2、代码实现

与前两种模式不同的时，我们在实现该模式的生产者时建立连接后需要去创建交换机，由于交换机不同的类型其工作模式不同，所以我们创建交换机时 需要去指定他的类型，创建完成后我们还需要去将邮箱队列与短信队列与该交换机进行绑定

创建交换机需要使用该方法exchangeDeclare

参数说明
exchange交换机名
type交换机类型
durable是否持久化到内存，一般为true
autoDelete是否自动删除，一般为false
internal内部使用参数，一般为false
arguments参数信息，一般为null

队列绑定交换机需要用到这个queueBind（队列名，交换机名，路由名--如果交换机类型为fanout则为空字符串）方法

生产者：

import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;import java.io.IOException;
import java.util.concurrent.TimeoutException;public class Producer {private static final String EXCHANGE_NAME = "pubSubExchange";private static final String EMAIL_QUEUE = "email";private static final String SMS_QUEUE = "sms";public static void main(String[] args) throws IOException, TimeoutException {// 1.创建连接工厂 配置连接工厂ConnectionFactory factory = new ConnectionFactory();factory.setHost("127.0.0.1");factory.setPort(5672);factory.setVirtualHost("/DemoVirtualHost");factory.setUsername("guest");factory.setPassword("guest");// 2.建立连接Connection connection = factory.newConnection();Channel channel = connection.createChannel();// 3.创建交换机channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.FANOUT,true,false,false,null);// 4.创建队列channel.queueDeclare(EMAIL_QUEUE,true,false,false,null);channel.queueDeclare(SMS_QUEUE,true,false,false,null);// 5.绑定交换机channel.queueBind(EMAIL_QUEUE,EXCHANGE_NAME,"");channel.queueBind(SMS_QUEUE,EXCHANGE_NAME,"");// 6.发送信息channel.basicPublish(EXCHANGE_NAME,"",null,"hello".getBytes());// 7.关闭连接channel.close();connection.close();}
}

运行查看服务器 

 此时两个消费者只需修改监听的队列即可

email消费者：

import com.rabbitmq.client.*;import java.io.IOException;
import java.util.concurrent.TimeoutException;public class EmailConsumer {private static final String EMAIL_QUEUE = "email";private static final String SMS_QUEUE = "sms";public static void main(String[] args) throws IOException, TimeoutException {// 1.创建连接工厂 配置工厂ConnectionFactory factory = new ConnectionFactory();factory.setHost("127.0.0.1");factory.setPort(5672);factory.setVirtualHost("/DemoVirtualHost");factory.setUsername("guest");factory.setPassword("guest");// 2.建立连接Connection connection = factory.newConnection();Channel channel = connection.createChannel();// 3.监听队列获取消息Consumer consumer = new DefaultConsumer(channel) {@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {System.out.println("邮件已发送" + new String(body));}};channel.basicConsume(EMAIL_QUEUE,true,consumer);}
}

运行代码 

sms消费者：

import com.rabbitmq.client.*;import java.io.IOException;
import java.util.concurrent.TimeoutException;public class EmailConsumer {private static final String EMAIL_QUEUE = "email";private static final String SMS_QUEUE = "sms";public static void main(String[] args) throws IOException, TimeoutException {// 1.创建连接工厂 配置工厂ConnectionFactory factory = new ConnectionFactory();factory.setHost("127.0.0.1");factory.setPort(5672);factory.setVirtualHost("/DemoVirtualHost");factory.setUsername("guest");factory.setPassword("guest");// 2.建立连接Connection connection = factory.newConnection();Channel channel = connection.createChannel();// 3.监听队列获取消息Consumer consumer = new DefaultConsumer(channel) {@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {System.out.println("已发送" + new String(body));}};channel.basicConsume(SMS_QUEUE,true,consumer);}
}

五、Routing路由模式

1、介绍

上述用户下单发送回购优惠券需求时，如果有的用户他注册时没有设置邮箱或者电话，我们不需要去给没有的邮箱发送，此时使用订阅模式是不可取的，使用路由模式是可行的，那么为什么路由模式可以呢，让我们来看。路由模式与订阅模式相同的是都需要使用交换机，不同的是当收到生产者的消息时，订阅模式会分发给每一个与之绑定的队列，而路由模式则不同，他会按照消息指定的路由去匹配到相应的队列从而发送给指定的队列

2、代码实现

由上述路由模式图，我们在发送消息前，需要先创建一个交换机且路由模式交换机类型不在是fanout而是direct路由模式，还需要去创建队列，然后绑定交换机，在绑定交换机的时候我们需要去给队列设置对应的路由，且在发送消息时指定路由

生产者：

import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;import java.io.IOException;
import java.util.concurrent.TimeoutException;public class Producer {private static final String EXCHANGE_NAME = "rout-ex";private static final String EMAIL_QUEUE = "email";private static final String SMS_EMAIL = "sms";public static void main(String[] args) throws IOException, TimeoutException {// 1.创建连接工厂 配置工厂ConnectionFactory factory = new ConnectionFactory();factory.setHost("127.0.0.1");factory.setPort(5672);factory.setVirtualHost("/DemoVirtualHost");factory.setUsername("guest");factory.setPassword("guest");// 2.建立连接Connection connection = factory.newConnection();Channel channel = connection.createChannel();// 3.创建交换机channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.DIRECT,true,false,false,null);// 4.创建队列channel.queueDeclare(EMAIL_QUEUE,true,false,false,null);channel.queueDeclare(SMS_EMAIL,true,false,false,null);// 5.绑定交换机channel.queueBind(EMAIL_QUEUE,EXCHANGE_NAME,"email");channel.queueBind(SMS_EMAIL,EXCHANGE_NAME,"sms");// 6.发送消息channel.basicPublish(EXCHANGE_NAME,"email",null,"routing".getBytes());// 7.关闭连接channel.close();connection.close();}
}

发送消息指定了路由为email，此时我们运行代码查看服务器

消费者：

修改监听队列名即可

import com.rabbitmq.client.*;import java.io.IOException;
import java.util.concurrent.TimeoutException;public class Consumer {public static void main(String[] args) throws IOException, TimeoutException {// 1.获取连接工厂ConnectionFactory connectionFactory = new ConnectionFactory();// 2.设置参数：队列、交换机、虚拟机、端口、IPconnectionFactory.setHost("127.0.0.1");connectionFactory.setPort(5672);connectionFactory.setVirtualHost("/DemoVirtualHost");connectionFactory.setUsername("1886");connectionFactory.setPassword("1886");// 3.建立连接 Connection ---- channelConnection connection = connectionFactory.newConnection();Channel channel = connection.createChannel();// 4.获取消息com.rabbitmq.client.Consumer consumer = new DefaultConsumer(channel) {@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {System.out.println("email消费");}};channel.basicConsume("email",true,consumer);}
}

 

六、Topics通配符模式

1、介绍

上述路由模式我们知道了消息可以指定路由进而到达指定的队列，但路由只有一个所以该消息只能进入一个队列，而后续如果有一个消息需要进入几个队列的需求时订阅模式与路由模式就不能完成，此时我们可以使用通配符模式进行匹配，而该模式的交换机类型为通配符类型的topic。

通配符有两种

种类介绍
*（system.*）*指的是匹配一个单词，（system.*）是指路由前缀为system.且后面有一个单词即可
#（#.error）#指的是匹配0个或多个单词，（#.error）是指路由前面没有或有若干单词，后面为.error

如果此时服务器有两个队列一个为存放所有分布式系统的error日志，一个只用于存放A系统的所有等级日志 ，此时前者路由配置为#.error，后者可为A.*

2、代码实现

生产者：

我们实现上述业务时可对队列进行上述路由配置，此时我们将消息路由设置为B系统下的error日志：B.error，该消息就会进入error队列

import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;import java.io.IOException;
import java.util.concurrent.TimeoutException;public class Producer {private static final String EXCHANGE_NAME = "topic-ex";private static final String ERROR_QUEUE = "error";private static final String SYSTEM_LOG = "a-system-logging";public static void main(String[] args) throws IOException, TimeoutException {// 1.创建连接工厂 配置工厂ConnectionFactory factory = new ConnectionFactory();factory.setHost("127.0.0.1");factory.setPort(5672);factory.setVirtualHost("/DemoVirtualHost");factory.setUsername("guest");factory.setPassword("guest");// 2.建立连接Connection connection = factory.newConnection();Channel channel = connection.createChannel();// 3.创建交换机channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.TOPIC,true,false,false,null);// 4.创建队列channel.queueDeclare(ERROR_QUEUE,true,false,false,null);channel.queueDeclare(SYSTEM_LOG,true,false,false,null);// 5.绑定交换机channel.queueBind(ERROR_QUEUE,EXCHANGE_NAME,"#.error");  // 所有系统的error日志channel.queueBind(SYSTEM_LOG,EXCHANGE_NAME,"A.*");       // A系统的所有日志// 6.发送消息channel.basicPublish(EXCHANGE_NAME,"B.error",null,"b系统的error日志".getBytes());//  7.关闭连接channel.close();connection.close();}
}

消费者：

消费者只需要修改监听的队列名即可

import com.rabbitmq.client.*;import java.io.IOException;
import java.util.concurrent.TimeoutException;public class Consumer {public static void main(String[] args) throws IOException, TimeoutException {// 1.创建连接工厂 配置工厂ConnectionFactory factory = new ConnectionFactory();factory.setHost("127.0.0.1");factory.setPort(5672);factory.setVirtualHost("/DemoVirtualHost");factory.setUsername("guest");factory.setPassword("guest");// 2.建立连接Connection connection = factory.newConnection();Channel channel = connection.createChannel();// 3.监听队列获取消息com.rabbitmq.client.Consumer consumer = new DefaultConsumer(channel){@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {System.out.println("处理日志" + new String(body));}};channel.basicConsume("error",true,consumer);}
}

 运行代码即可查看

http://www.jmfq.cn/news/5137183.html

相关文章：

  • 咸阳做网站的公司/百度金融
  • 国外做项目的网站/吉林关键词排名优化软件
  • 兰溪市城乡建设局网站/基本seo技术在线咨询
  • wordpress网站绑定多个域名/seo关键词智能排名
  • 怎么做自己的公司网站放在百度/常见的搜索引擎有哪些
  • 网站开发怎么开发/谷歌优化怎么做
  • 顺企网查企业电话/关键词优化公司哪家效果好
  • 如何给网站数据备份/优化大师是什么软件
  • 如何成功开展网络营销/seo关键词排名优化官网
  • 福州网站推广优化/网址查询站长工具
  • 做网站用服务器/电话销售怎么找客户渠道
  • 临沂网站建设哪家公司好/推广软文模板
  • 河南和城乡建设厅网站/今天刚刚发生的新闻最新新闻
  • 可靠的广州做网站/微信crm
  • 深圳网站建设公司推荐/求老哥给几个靠谱的网站
  • 免费ppt模板下载大全 完整版无需会员/网站关键词搜索排名优化
  • 厦门 网站备案/百度指数怎么下载
  • joomla建站教程/湖南企业竞价优化
  • 我想出租做房 请问哪个网站好些/seo专业培训学费多少钱
  • 做网站效果图/哈尔滨seo公司
  • 中国建设官方网站登录/厦门百度竞价推广
  • wordpress可以做门户网站/百度快速收录权限域名
  • 怎么做网站规划书/嘉兴seo计费管理
  • 国外的一些网站/互联网去哪里学
  • 新闻网站如何做推广/网站优化内容
  • 网站首页优化/seo和sem的区别是什么
  • 信誉好的营销网站建设/三只松鼠搜索引擎营销案例
  • 个人做啥网站流量大/nba最新排行
  • 以前的网站忘了怎么办啊/app推广
  • 门头沟网站建设/seo咨询常德