B站學(xué)習(xí)RabbitMQ筆記圖靈1、使用場景2、安裝RabbitMQ3、RabbitMQ工作原理4、direct交換機(jī)5、fanout交換機(jī)6、topic交換機(jī)1、使用場景例如有一個(gè)推送數(shù)據(jù)到第三方接口的場景可以把數(shù)據(jù)先推送到消息隊(duì)列中消費(fèi)者接口消息并執(zhí)行業(yè)務(wù)邏輯推送第三方接口根據(jù)返回結(jié)果決定是ack確認(rèn)消息還是重試/進(jìn)死信隊(duì)列。一定程度上網(wǎng)絡(luò)可以做到消息必達(dá)第三方。2、安裝RabbitMQ需要安裝RabbitMQ并運(yùn)行生產(chǎn)者和消費(fèi)者都連接到RabbitMQ生產(chǎn)推送消息到RabbitMQ消費(fèi)者從RabbitMQ中接收消息?？梢园裄abbitMQ安裝到docker中不用配置環(huán)境docker run-d \--name rabbitmq \-p 5672:5672 \-p 15672:15672 \-e RABBITMQ_DEFAULT_USERadmin \-e RABBITMQ_DEFAULT_PASSadmin123 \ rabbitmq:management啟動(dòng)后管理界面地址為 http://ip:15672賬號(hào)密碼分別為 admin / admin123。如果訪問不進(jìn)去就去阿里云開放端口3、RabbitMQ工作原理4、direct交換機(jī)這個(gè)交換機(jī)是根據(jù)routingKey去找到對應(yīng)的綁定queue如果多個(gè)queue綁定了同一個(gè)routingKey那么推送routingKey消息時(shí)會(huì)推動(dòng)到對應(yīng)多個(gè)queue觀察生成者代碼交換機(jī)和隊(duì)列綁定生產(chǎn)者packagecom.example.rabbitmq1.demos.rabbitmq;importcom.rabbitmq.client.BuiltinExchangeType;importcom.rabbitmq.client.Channel;importcom.rabbitmq.client.Connection;importcom.rabbitmq.client.ConnectionFactory;importjava.io.IOException;importjava.util.concurrent.TimeoutException;/** * author daxia * date 2026/10/12 * description rabbitmq生產(chǎn)者 */publicclassProducerDirect{publicstaticvoidmain(String[]args)throwsIOException,TimeoutException{StringqueueName1mq_queue_name1;StringqueueName2mq_queue_name2;StringqueueName3mq_queue_name3;StringqueueName4mq_queue_name4;//路由key,StringroutingKey1routingKey1;StringroutingKey2routingKey2;StringroutingKey3routingKey3;StringroutingKey4routingKey4;StringexchangeNamemq_exchange_name;//創(chuàng)建一個(gè)連接工廠ConnectionFactoryfactorynewConnectionFactory();//mq服務(wù)端ipfactory.setHost(8.138.123.9);//mq賬號(hào)factory.setUsername(admin);//mq密碼factory.setPassword(admin123);//端口factory.setPort(5672);//創(chuàng)建連接Connectionconnectionfactory.newConnection();//創(chuàng)建信道Channelchannelconnection.createChannel();//創(chuàng)建交換機(jī)channel.exchangeDeclare(exchangeName,BuiltinExchangeType.DIRECT,true,false,null);//生成四個(gè)隊(duì)列channel.queueDeclare(queueName1,true,false,false,null);channel.queueDeclare(queueName2,true,false,false,null);channel.queueDeclare(queueName3,true,false,false,null);channel.queueDeclare(queueName4,true,false,false,null);//隊(duì)列和交換機(jī)綁定 key1 綁定三個(gè)隊(duì)列涂松key1消息時(shí)三個(gè)隊(duì)列都會(huì)收到消息channel.queueBind(queueName1,exchangeName,routingKey1);channel.queueBind(queueName2,exchangeName,routingKey1);channel.queueBind(queueName3,exchangeName,routingKey1);channel.queueBind(queueName4,exchangeName,routingKey4);//發(fā)送消息Stringmsg1做完這個(gè)案例就可以休息了1;Stringmsg2做完這個(gè)案例就可以休息了2;Stringmsg3做完這個(gè)案例就可以休息了3;Stringmsg4做完這個(gè)案例就可以休息了4;//channel.basicPublish(exchangeName,routingKey1,null,msg1.getBytes());channel.basicPublish(exchangeName,routingKey2,null,msg2.getBytes());channel.basicPublish(exchangeName,routingKey3,null,msg3.getBytes());channel.basicPublish(exchangeName,routingKey4,null,msg4.getBytes());channel.close();connection.close();}}消費(fèi)者packagecom.example.rabbitmq3.rabbitmq;importcom.rabbitmq.client.*;importjava.io.IOException;importjava.util.concurrent.TimeoutException;/** * author daxia * date 2026/10/12 * description RabbitMQ消費(fèi)者Direct */publicclassComsumerDirect{publicstaticvoidmain(String[]args)throwsIOException,TimeoutException{//創(chuàng)建一個(gè)連接工廠ConnectionFactoryfactorynewConnectionFactory();//mq服務(wù)端ipfactory.setHost(8.138.123.9);//mq賬號(hào)factory.setUsername(admin);//mq密碼factory.setPassword(admin123);//端口factory.setPort(5672);//創(chuàng)建連接Connectionconnectionfactory.newConnection();//創(chuàng)建信道Channelchannelconnection.createChannel();DeliverCallbackdeliverCallback(consumerTag,message)-{System.out.println(接收到消息newString(message.getBody()));};CancelCallbackcancelCallbackconsumerTag-{System.out.println(消息中斷);};//消費(fèi)消息channel.basicConsume(mq_queue_name1,true,deliverCallback,cancelCallback);channel.basicConsume(mq_queue_name2,true,deliverCallback,cancelCallback);channel.basicConsume(mq_queue_name3,true,deliverCallback,cancelCallback);channel.basicConsume(mq_queue_name4,true,deliverCallback,cancelCallback);}}5、fanout交換機(jī)這個(gè)交換機(jī)是群發(fā)會(huì)把消息推送到和當(dāng)前交換機(jī)綁定的所有隊(duì)列并且routingKey會(huì)失效。生產(chǎn)者packagecom.example.rabbitmq1.demos.rabbitmq;importcom.rabbitmq.client.BuiltinExchangeType;importcom.rabbitmq.client.Channel;importcom.rabbitmq.client.Connection;importcom.rabbitmq.client.ConnectionFactory;importjava.io.IOException;importjava.util.concurrent.TimeoutException;/** * author daxia * date 2026/10/12 * description rabbitmq生產(chǎn)者 */publicclassProducerFanout{publicstaticvoidmain(String[]args)throwsIOException,TimeoutException{StringqueueName1mq_queue_fanout_name1;StringqueueName2mq_queue_fanout_name2;StringqueueName3mq_queue_fanout_name3;StringqueueName4mq_queue_fanout_name4;//路由key,StringroutingKey1routingKey1;StringroutingKey2routingKey2;StringroutingKey3routingKey3;StringroutingKey4routingKey4;StringexchangeNamemq__fanout_exchange_name;//創(chuàng)建一個(gè)連接工廠ConnectionFactoryfactorynewConnectionFactory();//mq服務(wù)端ipfactory.setHost(8.138.123.9);//mq賬號(hào)factory.setUsername(admin);//mq密碼factory.setPassword(admin123);//端口factory.setPort(5672);//創(chuàng)建連接Connectionconnectionfactory.newConnection();//創(chuàng)建信道Channelchannelconnection.createChannel();//創(chuàng)建交換機(jī) FANOUT 是群發(fā)所有和交換機(jī)綁定的隊(duì)列都能接收到消息 routingKey無效channel.exchangeDeclare(exchangeName,BuiltinExchangeType.FANOUT,true,false,null);//生成四個(gè)隊(duì)列channel.queueDeclare(queueName1,true,false,false,null);channel.queueDeclare(queueName2,true,false,false,null);channel.queueDeclare(queueName3,true,false,false,null);channel.queueDeclare(queueName4,true,false,false,null);//隊(duì)列和交換機(jī)綁定 key1 綁定三個(gè)隊(duì)列涂松key1消息時(shí)三個(gè)隊(duì)列都會(huì)收到消息channel.queueBind(queueName1,exchangeName,routingKey1);channel.queueBind(queueName2,exchangeName,routingKey1);channel.queueBind(queueName3,exchangeName,routingKey1);channel.queueBind(queueName4,exchangeName,routingKey4);//發(fā)送消息Stringmsg1Fanout類型的交換機(jī)會(huì)把消息發(fā)送給所有與之綁定的隊(duì)列;Stringmsg2做完這個(gè)案例就可以休息了2;Stringmsg3做完這個(gè)案例就可以休息了3;Stringmsg4做完這個(gè)案例就可以休息了4;//channel.basicPublish(exchangeName,routingKey1,null,msg1.getBytes());channel.close();connection.close();}}消費(fèi)者packagecom.example.rabbitmq3.rabbitmq;importcom.rabbitmq.client.*;importjava.io.IOException;importjava.util.concurrent.TimeoutException;/** * author daxia * date 2026/10/12 * description RabbitMQ消費(fèi)者Fanout */publicclassComsumerFanout{publicstaticvoidmain(String[]args)throwsIOException,TimeoutException{//創(chuàng)建一個(gè)連接工廠ConnectionFactoryfactorynewConnectionFactory();//mq服務(wù)端ipfactory.setHost(8.138.123.9);//mq賬號(hào)factory.setUsername(admin);//mq密碼factory.setPassword(admin123);//端口factory.setPort(5672);//創(chuàng)建連接Connectionconnectionfactory.newConnection();//創(chuàng)建信道Channelchannelconnection.createChannel();DeliverCallbackdeliverCallback(consumerTag,message)-{System.out.println(接收到消息newString(message.getBody()));};CancelCallbackcancelCallbackconsumerTag-{System.out.println(消息中斷);};//消費(fèi)消息channel.basicConsume(mq_queue_fanout_name1,true,deliverCallback,cancelCallback);channel.basicConsume(mq_queue_fanout_name2,true,deliverCallback,cancelCallback);channel.basicConsume(mq_queue_fanout_name3,true,deliverCallback,cancelCallback);channel.basicConsume(mq_queue_fanout_name4,true,deliverCallback,cancelCallback);}}6、topic交換機(jī)匹配規(guī)則routing key 和 binding key 都是由 . 分隔的單詞列表例如 order.pay.success、user.login。binding key 中支持兩個(gè)通配符*匹配恰好一個(gè)單詞#匹配零個(gè)或多個(gè)單詞匹配是整個(gè)字符串的單詞級(jí)匹配不是子串匹配。大小寫敏感。通配符通常要作為獨(dú)立單詞使用例如.orange.、lazy.#、#.rabbit。例如routingKey1.routingKey2.routingKey3.routingKey4就能匹配上代碼中的四個(gè)routingKey生產(chǎn)者packagecom.example.rabbitmq1.demos.rabbitmq;importcom.rabbitmq.client.BuiltinExchangeType;importcom.rabbitmq.client.Channel;importcom.rabbitmq.client.Connection;importcom.rabbitmq.client.ConnectionFactory;importjava.io.IOException;importjava.util.concurrent.TimeoutException;/** * author daxia * date 2026/10/12 * description rabbitmq生產(chǎn)者 */publicclassProducerTopic{publicstaticvoidmain(String[]args)throwsIOException,TimeoutException{StringqueueName1mq_queue_topic_name1;StringqueueName2mq_queue_topic_name2;StringqueueName3mq_queue_topic_name3;StringqueueName4mq_queue_topic_name4;//路由key,StringroutingKey1routingKey1.#;StringroutingKey2*.routingKey2.#;StringroutingKey3#.routingKey3.*;StringroutingKey4#.routingKey4;StringexchangeNamemq__topic_exchange_name;//創(chuàng)建一個(gè)連接工廠ConnectionFactoryfactorynewConnectionFactory();//mq服務(wù)端ipfactory.setHost(8.138.123.9);//mq賬號(hào)factory.setUsername(admin);//mq密碼factory.setPassword(admin123);//端口factory.setPort(5672);//創(chuàng)建連接Connectionconnectionfactory.newConnection();//創(chuàng)建信道Channelchannelconnection.createChannel();//創(chuàng)建交換機(jī) Topic交換機(jī)匹配隊(duì)列時(shí)是模糊匹配有點(diǎn)類似正則表達(dá)式channel.exchangeDeclare(exchangeName,BuiltinExchangeType.TOPIC,true,false,null);//生成四個(gè)隊(duì)列channel.queueDeclare(queueName1,true,false,false,null);channel.queueDeclare(queueName2,true,false,false,null);channel.queueDeclare(queueName3,true,false,false,null);channel.queueDeclare(queueName4,true,false,false,null);//隊(duì)列和交換機(jī)綁定 key1 綁定三個(gè)隊(duì)列涂松key1消息時(shí)三個(gè)隊(duì)列都會(huì)收到消息channel.queueBind(queueName1,exchangeName,routingKey1);channel.queueBind(queueName2,exchangeName,routingKey1);channel.queueBind(queueName3,exchangeName,routingKey1);channel.queueBind(queueName4,exchangeName,routingKey4);//發(fā)送消息Stringmsg1做完這個(gè)案例就可以休息了1;Stringmsg2做完這個(gè)案例就可以休息了2;Stringmsg3做完這個(gè)案例就可以休息了3;Stringmsg4做完這個(gè)案例就可以休息了4;//channel.basicPublish(exchangeName,routingKey1.routingKey2.routingKey3.routingKey4,null,msg1.getBytes());channel.close();connection.close();}}消費(fèi)者packagecom.example.rabbitmq3.rabbitmq;importcom.rabbitmq.client.*;importjava.io.IOException;importjava.util.concurrent.TimeoutException;/** * author daxia * date 2026/10/12 * description RabbitMQ消費(fèi)者Topic */publicclassComsumerTopic{publicstaticvoidmain(String[]args)throwsIOException,TimeoutException{//創(chuàng)建一個(gè)連接工廠ConnectionFactoryfactorynewConnectionFactory();//mq服務(wù)端ipfactory.setHost(8.138.123.9);//mq賬號(hào)factory.setUsername(admin);//mq密碼factory.setPassword(admin123);//端口factory.setPort(5672);//創(chuàng)建連接Connectionconnectionfactory.newConnection();//創(chuàng)建信道Channelchannelconnection.createChannel();DeliverCallbackdeliverCallback(consumerTag,message)-{System.out.println(接收到消息newString(message.getBody()));};CancelCallbackcancelCallbackconsumerTag-{System.out.println(消息中斷);};//消費(fèi)消息channel.basicConsume(mq_queue_topic_name1,true,deliverCallback,cancelCallback);channel.basicConsume(mq_queue_topic_name2,true,deliverCallback,cancelCallback);channel.basicConsume(mq_queue_topic_name3,true,deliverCallback,cancelCallback);channel.basicConsume(mq_queue_topic_name4,true,deliverCallback,cancelCallback);}}