网站seo在哪里设置,建设好网站,旅游网站系统设计,服务器放多个网站Hi~#xff01;这里是奋斗的明志#xff0c;很荣幸您能阅读我的文章#xff0c;诚请评论指点#xff0c;欢迎欢迎 ~~ #x1f331;#x1f331;个人主页#xff1a;奋斗的明志 #x1f331;#x1f331;所属专栏#xff1a;RabbitMQ #x1f4da;本系列文章为个人学… Hi~这里是奋斗的明志很荣幸您能阅读我的文章诚请评论指点欢迎欢迎 ~~ 个人主页奋斗的明志 所属专栏RabbitMQ 本系列文章为个人学习笔记在这里撰写成文一为巩固知识二为展示我的学习过程及理解。文笔、排版拙劣望见谅。 发布订阅模式、路由模式、通配符模式 一、Publish/Subscribe(发布/订阅)1、引入依赖2、编写配置类3、编写生产者代码4、编写消费者代码 二、Routing (路由模式)1、引入依赖2、编写配置类3、编写生产者代码4、编写消费者代码 三、Topics (通配符模式)1、引入依赖2、编写配置类3、编写生产者代码4、编写消费者代码 一、Publish/Subscribe(发布/订阅) 在发布/订阅模型中多了一个Exchange角色. Exchange 常见有三种类型, 分别代表不同的路由规则 a) Fanout:广播将消息交给所有绑定到交换机的队列 (Publish/Subscribe模式) b) Direct:定向把消息交给符合指定routing key的队列(Routing模式) c) Topic:通配符把消息交给符合routing pattern(路由模式)的队列(Topics模式) 也就分别对应不同的工作模式 1、引入依赖 dependencies!-- https://mvnrepository.com/artifact/com.rabbitmq/amqp-client --dependencygroupIdcom.rabbitmq/groupIdartifactIdamqp-client/artifactIdversion5.20.0/version/dependency
/dependencies2、编写配置类
package rabbitmq.constant;public class Constants {public static final String HOST 123.57.16.61;public static final Integer PORT 5672;public static final String USERNAME study;public static final String PASSWORD study;public static final String VIRTUAL_HOST bite;//发布订阅模式public static final String FANOUT_EXCHANGE fanout.exchange;public static final String FANOUT_QUEUE1 fanout.queue1;public static final String FANOUT_QUEUE2 fanout.queue2;
} 3、编写生产者代码
package rabbitmq.fanout;import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import rabbitmq.constant.Constants;import java.io.IOException;
import java.util.concurrent.TimeoutException;public class Producer {public static void main(String[] args) throws IOException, TimeoutException {ConnectionFactory connectionFactory new ConnectionFactory();connectionFactory.setHost(Constants.HOST);connectionFactory.setPort(Constants.PORT);connectionFactory.setUsername(Constants.USERNAME);connectionFactory.setPassword(Constants.PASSWORD);connectionFactory.setVirtualHost(Constants.VIRTUAL_HOST);Connection connection connectionFactory.newConnection();Channel channel connection.createChannel();//声明交换机/*** 交换机名称交换机类型开启可持久化关机数据不会丢失*/channel.exchangeDeclare(Constants.FANOUT_EXCHANGE, BuiltinExchangeType.FANOUT, true);//声明队列//queueDeclare 队列声明channel.queueDeclare(Constants.FANOUT_QUEUE1, true, false, false, null);channel.queueDeclare(Constants.FANOUT_QUEUE2, true, false, false, null);//交换机和队列进行绑定channel.queueBind(Constants.FANOUT_QUEUE1, Constants.FANOUT_EXCHANGE, );channel.queueBind(Constants.FANOUT_QUEUE2, Constants.FANOUT_EXCHANGE, );//发布消息String msg hello fanout...;//basicPublish (基础发布)channel.basicPublish(Constants.FANOUT_EXCHANGE, , null, msg.getBytes());System.out.println(消息发送成功);//关闭资源channel.close();connection.close();}
} 点击运行 4、编写消费者代码
消费者1
package rabbitmq.fanout;import com.rabbitmq.client.*;
import rabbitmq.constant.Constants;import java.io.IOException;
import java.util.concurrent.TimeoutException;public class Consumer1 {public static void main(String[] args) throws IOException, TimeoutException {ConnectionFactory connectionFactory new ConnectionFactory();connectionFactory.setHost(Constants.HOST);connectionFactory.setPort(Constants.PORT);connectionFactory.setUsername(Constants.USERNAME);connectionFactory.setPassword(Constants.PASSWORD);connectionFactory.setVirtualHost(Constants.VIRTUAL_HOST);Connection connection connectionFactory.newConnection();Channel channel connection.createChannel();//声明队列//queueDeclare 队列声明 (也可以省略)channel.queueDeclare(Constants.FANOUT_QUEUE1, true, false, false, null);//消费消息DefaultConsumer 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(Constants.FANOUT_QUEUE1,true,consumer);}
}
消费者2
package rabbitmq.fanout;import com.rabbitmq.client.*;
import rabbitmq.constant.Constants;import java.io.IOException;
import java.util.concurrent.TimeoutException;public class Consumer2 {public static void main(String[] args) throws IOException, TimeoutException {ConnectionFactory connectionFactory new ConnectionFactory();connectionFactory.setHost(Constants.HOST);connectionFactory.setPort(Constants.PORT);connectionFactory.setUsername(Constants.USERNAME);connectionFactory.setPassword(Constants.PASSWORD);connectionFactory.setVirtualHost(Constants.VIRTUAL_HOST);Connection connection connectionFactory.newConnection();Channel channel connection.createChannel();//声明队列//queueDeclare 队列声明 (也可以省略)channel.queueDeclare(Constants.FANOUT_QUEUE2, true, false, false, null);//消费消息DefaultConsumer 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(Constants.FANOUT_QUEUE2,true,consumer);}
} 二、Routing (路由模式) 队列和交换机的绑定, 不能是任意的绑定了, 而是要指定⼀个BindingKey(RoutingKey的⼀种) 消息的发送方在向 Exchange 发送消息时, 也需要指定消息的 RoutingKey Exchange也不再把消息交给每⼀个绑定的key, 而是根据消息的RoutingKey进行判断, 只有队列绑定时的BindingKey和发送消息的RoutingKey 完全⼀致, 才会接收到消息 1、引入依赖 dependencies!-- https://mvnrepository.com/artifact/com.rabbitmq/amqp-client --dependencygroupIdcom.rabbitmq/groupIdartifactIdamqp-client/artifactIdversion5.20.0/version/dependency
/dependencies2、编写配置类
package rabbitmq.constant;public class Constants {public static final String HOST 123.57.16.61;public static final Integer PORT 5672;public static final String USERNAME study;public static final String PASSWORD study;public static final String VIRTUAL_HOST bite;//路由模式public static final String DIRECT_EXCHANGE direct.exchange;public static final String DIRECT_QUEUE1 direct.queue1;public static final String DIRECT_QUEUE2 direct.queue2;
}
3、编写生产者代码
package rabbitmq.direct;import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import rabbitmq.constant.Constants;import java.io.IOException;
import java.util.concurrent.TimeoutException;public class Producer {public static void main(String[] args) throws IOException, TimeoutException {ConnectionFactory connectionFactory new ConnectionFactory();connectionFactory.setHost(Constants.HOST);connectionFactory.setPort(Constants.PORT);connectionFactory.setUsername(Constants.USERNAME);connectionFactory.setPassword(Constants.PASSWORD);connectionFactory.setVirtualHost(Constants.VIRTUAL_HOST);Connection connection connectionFactory.newConnection();Channel channel connection.createChannel();//声明交换机/*** 交换机名称交换机类型开启可持久化关机数据不会丢失*/channel.exchangeDeclare(Constants.DIRECT_EXCHANGE, BuiltinExchangeType.DIRECT, true);//声明队列//queueDeclare 队列声明channel.queueDeclare(Constants.DIRECT_QUEUE1, true, false, false, null);channel.queueDeclare(Constants.DIRECT_QUEUE2, true, false, false, null);//交换机和队列进行绑定channel.queueBind(Constants.DIRECT_QUEUE1, Constants.DIRECT_EXCHANGE, a);channel.queueBind(Constants.DIRECT_QUEUE2, Constants.DIRECT_EXCHANGE, a);channel.queueBind(Constants.DIRECT_QUEUE2, Constants.DIRECT_EXCHANGE, b);channel.queueBind(Constants.DIRECT_QUEUE2, Constants.DIRECT_EXCHANGE, c);//发布消息String msg_a hello direct, my routingkey is a...;//basicPublish (基础发布)channel.basicPublish(Constants.DIRECT_EXCHANGE, , null, msg_a.getBytes());String msg_b hello direct, my routingkey is b...;channel.basicPublish(Constants.DIRECT_EXCHANGE, , null, msg_b.getBytes());String msg_c hello direct, my routingkey is c...;channel.basicPublish(Constants.DIRECT_EXCHANGE, , null, msg_c.getBytes());System.out.println(消息发送成功);//关闭资源channel.close();connection.close();}
} 4、编写消费者代码
消费者1
package rabbitmq.direct;import com.rabbitmq.client.*;
import rabbitmq.constant.Constants;import java.io.IOException;
import java.util.concurrent.TimeoutException;public class Consumer1 {public static void main(String[] args) throws IOException, TimeoutException {ConnectionFactory connectionFactory new ConnectionFactory();connectionFactory.setHost(Constants.HOST);connectionFactory.setPort(Constants.PORT);connectionFactory.setUsername(Constants.USERNAME);connectionFactory.setPassword(Constants.PASSWORD);connectionFactory.setVirtualHost(Constants.VIRTUAL_HOST);Connection connection connectionFactory.newConnection();Channel channel connection.createChannel();//声明队列//queueDeclare 队列声明 (也可以省略)channel.queueDeclare(Constants.DIRECT_QUEUE1, true, false, false, null);//消费消息DefaultConsumer 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(Constants.DIRECT_QUEUE1,true,consumer);}
} 消费者2
package rabbitmq.direct;import com.rabbitmq.client.*;
import rabbitmq.constant.Constants;import java.io.IOException;
import java.util.concurrent.TimeoutException;public class Consumer2 {public static void main(String[] args) throws IOException, TimeoutException {ConnectionFactory connectionFactory new ConnectionFactory();connectionFactory.setHost(Constants.HOST);connectionFactory.setPort(Constants.PORT);connectionFactory.setUsername(Constants.USERNAME);connectionFactory.setPassword(Constants.PASSWORD);connectionFactory.setVirtualHost(Constants.VIRTUAL_HOST);Connection connection connectionFactory.newConnection();Channel channel connection.createChannel();//声明队列//queueDeclare 队列声明 (也可以省略)channel.queueDeclare(Constants.DIRECT_QUEUE2, true, false, false, null);//消费消息DefaultConsumer 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(Constants.DIRECT_QUEUE2,true,consumer);}
} 三、Topics (通配符模式)
Topics 和Routing模式的区别是:
topics 模式使用的交换机类型为topic(Routing模式使用的交换机类型为direct)topic 类型的交换机在匹配规则上进行了扩展, Binding Key⽀持通配符匹配(direct类型的交换机路由规则是BindingKey和RoutingKey完全匹配) 在topic类型的交换机在匹配规则上, 有些要求:
RoutingKey 是⼀系列由点( . )分隔的单词, ⽐如 stock.usd.nyse , nyse.vmw , quick.orange.rabbit BindingKey 和RoutingKey⼀样, 也是点( . )分割的字符串.Binding Key中可以存在两种特殊字符串, 用于模糊匹配 * 表⽰⼀个单词 # 表⽰多个单词(0-N个)比如: • Binding Key 为d.a.b 会同时路由到Q1 和Q2 • Binding Key 为d.a.f 会路由到Q1 • Binding Key 为c.e.f 会路由到Q2 • Binding Key 为d.b.f 会被丢弃, 或者返回给⽣产者(需要设置mandatory参数)
1、引入依赖 dependencies!-- https://mvnrepository.com/artifact/com.rabbitmq/amqp-client --dependencygroupIdcom.rabbitmq/groupIdartifactIdamqp-client/artifactIdversion5.20.0/version/dependency
/dependencies2、编写配置类
package rabbitmq.constant;public class Constants {public static final String HOST 123.57.16.61;public static final Integer PORT 5672;public static final String USERNAME study;public static final String PASSWORD study;public static final String VIRTUAL_HOST bite;//通配符模式public static final String TOPIC_EXCHANGE topic.exchange;public static final String TOPIC_QUEUE1 topic.queue1;public static final String TOPIC_QUEUE2 topic.queue2;3、编写生产者代码
package rabbitmq.topic;import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import rabbitmq.constant.Constants;import java.io.IOException;
import java.util.concurrent.TimeoutException;public class Producer {public static void main(String[] args) throws IOException, TimeoutException {ConnectionFactory connectionFactory new ConnectionFactory();connectionFactory.setHost(Constants.HOST);connectionFactory.setPort(Constants.PORT);connectionFactory.setUsername(Constants.USERNAME);connectionFactory.setPassword(Constants.PASSWORD);connectionFactory.setVirtualHost(Constants.VIRTUAL_HOST);Connection connection connectionFactory.newConnection();Channel channel connection.createChannel();//声明交换机/*** 交换机名称交换机类型开启可持久化关机数据不会丢失*/channel.exchangeDeclare(Constants.TOPIC_EXCHANGE, BuiltinExchangeType.TOPIC, true);//声明队列//queueDeclare 队列声明channel.queueDeclare(Constants.TOPIC_QUEUE1, true, false, false, null);channel.queueDeclare(Constants.TOPIC_QUEUE2, true, false, false, null);//交换机和队列进行绑定channel.queueBind(Constants.TOPIC_QUEUE1, Constants.TOPIC_EXCHANGE, *.a.*);channel.queueBind(Constants.TOPIC_QUEUE2, Constants.TOPIC_EXCHANGE, *.*.b);channel.queueBind(Constants.TOPIC_QUEUE2, Constants.TOPIC_EXCHANGE, c.#);//发布消息String msg_a hello topic, my routingkey is ae.a.f...;//basicPublish (基础发布)channel.basicPublish(Constants.TOPIC_EXCHANGE, ae.a.f, null, msg_a.getBytes());String msg_b hello topic, my routingkey is ef.a.b...;channel.basicPublish(Constants.TOPIC_EXCHANGE, ef.a.b, null, msg_b.getBytes());String msg_c hello topic, my routingkey is c.ef.b...;channel.basicPublish(Constants.TOPIC_EXCHANGE, c.ef.b, null, msg_c.getBytes());System.out.println(消息发送成功);//关闭资源channel.close();connection.close();}
}
4、编写消费者代码
消费者1
package rabbitmq.topic;import com.rabbitmq.client.*;
import rabbitmq.constant.Constants;import java.io.IOException;
import java.util.concurrent.TimeoutException;public class Consumer1 {public static void main(String[] args) throws IOException, TimeoutException {ConnectionFactory connectionFactory new ConnectionFactory();connectionFactory.setHost(Constants.HOST);connectionFactory.setPort(Constants.PORT);connectionFactory.setUsername(Constants.USERNAME);connectionFactory.setPassword(Constants.PASSWORD);connectionFactory.setVirtualHost(Constants.VIRTUAL_HOST);Connection connection connectionFactory.newConnection();Channel channel connection.createChannel();//声明队列//queueDeclare 队列声明 (也可以省略)channel.queueDeclare(Constants.TOPIC_QUEUE1, true, false, false, null);//消费消息DefaultConsumer 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(Constants.TOPIC_QUEUE1,true,consumer);}
} 消费者2
package rabbitmq.topic;import com.rabbitmq.client.*;
import rabbitmq.constant.Constants;import java.io.IOException;
import java.util.concurrent.TimeoutException;public class Consumer2 {public static void main(String[] args) throws IOException, TimeoutException {ConnectionFactory connectionFactory new ConnectionFactory();connectionFactory.setHost(Constants.HOST);connectionFactory.setPort(Constants.PORT);connectionFactory.setUsername(Constants.USERNAME);connectionFactory.setPassword(Constants.PASSWORD);connectionFactory.setVirtualHost(Constants.VIRTUAL_HOST);Connection connection connectionFactory.newConnection();Channel channel connection.createChannel();//声明队列//queueDeclare 队列声明 (也可以省略)channel.queueDeclare(Constants.TOPIC_QUEUE2, true, false, false, null);//消费消息DefaultConsumer 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(Constants.TOPIC_QUEUE2,true,consumer);}
}
