举例：https://blog.csdn.net/qq_35387940/article/details/100514134
Windows下RabbitMQ安装及配置地址: https://blog.csdn.net/zhm3023/article/details/82217222RabbitMQ(四)订阅模式：https://blog.csdn.net/saytime/article/details/80541328RabbitMQ(五)路由模式：https://blog.csdn.net/saytime/article/details/80541402RabbitMQ(六)主题模式：https://blog.csdn.net/saytime/article/details/80541412
1、简单队列不足：即一个生产者对应一个消费者，一对一的关系，不支持多个消费者
2、工作队列模式：即一个生产者可以对应多个消费者同时消费，相比简单队列支持多消费者。
3、订阅模式：即一个生产者发送消息给多个消费者，且每个消费者都收到一次，也即是一个消息能够被多个消费者消费。类似于我们订阅同一微信公众号，微信公众号推送图文，我们每个人都能收到一份。
4、路由模式：跟订阅模式类似，只不过在订阅模式的基础上加上了类型，订阅模式是分发到所有绑定到交换机的队列，路由模式只分发到绑定在交换机上面指定路由键的队列。
5、主题模式：跟路由模式类似，只不过路由模式是指定固定的路由键，而主题模式是可以模糊匹配路由键，类似于SQL中=和like的关系。
//简单模式：// 获取连接.从连接开一个通道，发送消息到队列。 消费者消费队列的消息。
//订阅模式：// 获取连接.从连接开一个通道.声明一个fanout分发交换机.发送消息到交换机。消费者绑定到交换机。消费者消费具体的队列的消息。
/** * 生产者 */public class Send { private static final String EXCHANGE_NAME = "test_exchange_fanout";public static void main(String[] args) throws IOException, TimeoutException { // 获取连接 Connection cOnnection= ConnectionUtil.getConnection(); // 从连接开一个通道 Channel channel = connection.createChannel(); // 声明一个fanout分发交换机 channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.FANOUT); String message = "hello, ps"; // 发送消息 channel.basicPublish(EXCHANGE_NAME, "", null, message.getBytes()); System.out.println(" [x] Sent \'" + message + "\'"); channel.close(); connection.close(); }}/** * 短信消费者 */public class Recv { // 短信队列 private static final String QUEUE_NAME = "test_queue_fanout_sms"; private static final String EXCHANGE_NAME = "test_exchange_fanout"; public static void main(String[] args) throws IOException, TimeoutException { // 获取连接 Connection cOnnection= ConnectionUtil.getConnection(); // 打开通道 Channel channel = connection.createChannel(); // 申明要消费的队列 channel.queueDeclare(QUEUE_NAME, false, false, false, null); // 绑定队列到交换机 channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, ""); // 这样RabbitMQ就会使得每个Consumer在同一个时间点最多处理一个Message。换句话说，在接收到该Consumer的ack前，他它不会将新的Message分发给它。 channel.basicQos(1); // 创建一个回调的消费者处理类 Consumer cOnsumer= new DefaultConsumer(channel) { @Override public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { // 接收到的消息 String message = new String(body); System.out.println(" [1] Received \'" + message + "\'"); try { Thread.sleep(1000); } catch (InterruptedException e) { e.printStackTrace(); } finally { System.out.println(" [1] done "); channel.basicAck(envelope.getDeliveryTag(), false); } } }; // 消费消息 channel.basicConsume(QUEUE_NAME, false, consumer); }}RabbitMQ消息确认机制之事务机制：
1.服务器异常数据丢失问题？
第一种： AMQP 事务处理机制；https://blog.csdn.net/saytime/article/details/80541423
txSelect 将当前channel设置为transaction模式channel.txCommit 提交当前事务channel.txRollback 事务回滚成功就是发送到服务器，失败可以捕获异常再重发。缺点：降低rabbirmq的性能。第二种： confirm模式 。信号确认或者失败回复：
https://blog.csdn.net/cj337083172/article/details/79730438
1.同步：提交消息后，等待服务端回复后才算确认，概括的简单吧。2.异步：使用监听方法，异步等待消息回调。控制方式开启confirm模式代码 : channel.confirmSelect();1.同步确认，等待结果方法： channel.waitForConfirms() true-发送成功，false-发送失败2.异步确认，直接添加一个监听，也能叫做监听模式：channel.addConfirmListener(new ConfirmListener() {//实现返回处理方法... });