การใช้มิดเดิลแวร์ข้อความสปริงบูต

1. คำนำ

มีแนวคิดที่สำคัญ 2 แนวคิดในมิดเดิลแวร์ข้อความ: นายหน้าข้อความและปลายทาง เมื่อผู้ส่งข้อความส่งข้อความข้อความจะถูกลบโดยนายหน้าข้อความซึ่งทำให้มั่นใจได้ว่าข้อความจะถูกส่งไปยังปลายทางที่ระบุ

โบรกเกอร์ข้อความที่ใช้กันทั่วไปของเรารวมถึงข้อกำหนด JMS และ AMQP ตามลําดับการใช้งานทั่วไปของพวกเขาคือ ActiveMQ และ RabbitMQ

2. รวม ActiveMQ

2.1 เพิ่มการพึ่งพา

 <การพึ่งพา> <roupId> org.springframework.boot </groupId> <ratifactid> Spring-Boot-Starter-ActiveMq </artifactid> </dermentrency> <!-หากคุณต้องการกำหนดค่าการเชื่อมต่อ </dermentency>

2.2 เพิ่มการกำหนดค่า

 # activemq กำหนดค่า spring.activemq.broker-url = tcp: //192.168.2.12: 61616Spring.activeMq.user = adminspring.activeMq.password = AdminSpring.ActiveMq.pool.enabled การกำหนดค่าจะต้องตั้งค่าเป็น truespring.jms.pub-sub-domain = false

ที่นี่ Spring.activeMq.pool.enabled = False หมายถึงการปิดกลุ่มการเชื่อมต่อ

2.3 การเข้ารหัส

คลาสการกำหนดค่า:

 @ConfigurationPublic คลาส JMSConfirguration {สาธารณะคงที่สาธารณะสุดท้าย queue_name = "Activemq_queue"; สตริงสุดท้ายคงที่สาธารณะ topic_name = "activeemq_topic"; @Bean สาธารณะคิวคิว () {ส่งคืนใหม่ activeMqqueue (queue_name); } @Bean หัวข้อสาธารณะหัวข้อ () {ส่งคืนใหม่ activeMqTopic (topic_name); -

รับผิดชอบในการสร้างคิวและหัวข้อ

ผู้ผลิตข้อความ:

 @ComponentPublic คลาส JMSSENDER {@AutoWired คิวคิวส่วนตัว; หัวข้อหัวข้อส่วนตัว @autowired; @autoWired JMSMESSAGINGTEMPLATE JMSTEMPLATE; โมฆะสาธารณะ SendByQueue (ข้อความสตริง) {this.jmstemplate.convertandsend (คิวข้อความ); } โมฆะสาธารณะ sendbytopic (ข้อความสตริง) {this.jmstemplate.convertandsend (หัวข้อ, ข้อความ); -

ผู้บริโภคข้อความ:

 @ComponentPublic คลาส JMSRECEIVER {@JMSLISTENER (ปลายทาง = JMSCONFIRGURATION.Queue_Name) โมฆะสาธารณะรับ BetByQueue (ข้อความสตริง) {System.out.println ("รับข้อความคิว:" + ข้อความ); } @jmslistener (ปลายทาง = jmsconfirguration.topic_name) โมฆะสาธารณะรับ bytopic (ข้อความสตริง) {system.out.println ("รับข้อความหัวข้อ:" + ข้อความ); -

ผู้บริโภคข้อความฟังข้อความโดยใช้คำอธิบายประกอบ @JMSListener

2.4 การทดสอบ

 @runwith (SpringRunner.class) @springboottestpublic คลาส JMSTEST {@AutoWired Private JMSSender ผู้ส่ง; @Test โมฆะสาธารณะ testSendByQueue () {สำหรับ (int i = 1; i <6; i ++) {this.sender.sendbyqueue ("สวัสดี activemq queue"+i); }} @Test โมฆะสาธารณะ testSendByTopic () {สำหรับ (int i = 1; i <6; i ++) {this.sender.sendbytopic ("สวัสดี activemq topic"+i); -

ผลการพิมพ์:

รับข้อความคิว: สวัสดี activemq คิว 1
รับข้อความคิว: สวัสดี activemq คิว 2
รับข้อความคิว: สวัสดี activemq คิว 3
รับข้อความคิว: สวัสดี activemq คิว 4
รับข้อความคิว: สวัสดี activemq คิว 5

ตั้งค่า spring.jms.pub-sub-domain = true เมื่อทดสอบโหมดการเผยแพร่/สมัครสมาชิก

รับข้อความหัวข้อ: สวัสดี activemq หัวข้อ 1
รับข้อความหัวข้อ: สวัสดี activemq หัวข้อ 2
รับข้อความหัวข้อ: สวัสดี activemq หัวข้อ 3
รับข้อความหัวข้อ: สวัสดี activemq หัวข้อ 4
รับข้อความหัวข้อ: สวัสดี activemq หัวข้อ 5

3. รวม RabbitMQ

3.1 เพิ่มการพึ่งพา

 <การพึ่งพา> <roupId> org.springframework.boot </groupId> <ratifactid> Spring-Boot-Starter-AMQP </artifactId>

3.2 เพิ่มการกำหนดค่า

 spring.rabbitmq.host = 192.168.2.30spring.rabbitmq.port = 5672spring.rabbitmq.username = Lightspring.rabbitmq.password = lightspring.rabbitmq.virtual-host =/ทดสอบ

3.3 การเข้ารหัส

คลาสการกำหนดค่า:

 @ConfigurationPublic คลาส AMQPCONFIRGURATION {// =================================================== สตริงสุดท้ายคงที่สาธารณะ @Bean สาธารณะคิวคิว () {ส่งคืนคิวใหม่ (simple_queue, true); } // ================== สตริงสุดท้ายคงที่สาธารณะ ps_queue_1 = "ps_queue_1"; สตริงสุดท้ายคงที่สาธารณะ ps_queue_2 = "ps_queue_2"; สตริงสุดท้ายคงที่ FANOUT_EXCHANGE = "FANOUT_EXCHANGE"; @Bean สาธารณะคิว psqueue1 () {ส่งคืนคิวใหม่ (ps_queue_1, จริง); } @Bean Public Queue Psqueue2 () {ส่งคืนคิวใหม่ (ps_queue_2, จริง); } @Bean Public FanoutExChange FanoutExchange () {ส่งคืน FanOutExchange ใหม่ (FANOUT_EXCHANGE); } @Bean Public Binding FanoutBinding1 () {return bindingBuilder.bind (psqueue1 ()). ถึง (fanoutExchange ()); } @Bean Public Binding FanoutBinding2 () {return bindingBuilder.bind (psqueue2 ()). ถึง (fanoutExchange ()); } // =========================== สตริงสุดท้ายคงที่สาธารณะเส้นทางสุดท้ายเส้นทาง _Queue_1 = "RAUTING_QUEUE_1"; สาธารณะคงที่สตริงสุดท้ายเส้นทางการกำหนดเส้นทาง _queue_2 = "routing_queue_2"; สตริงสุดท้ายคงที่สาธารณะ direct_exchange = "direct_exchange"; @Bean Public Queue RoutingQueueUE1 () {ส่งคืนคิวใหม่ (เส้นทาง _queue_1, จริง); } @Bean RoutingQueueUEUEUEUE2 () {ส่งคืนคิวใหม่ (routing_queue_2, จริง); } @Bean Public DirectExChange DirectExChange () {ส่งคืน DirectExChange ใหม่ (direct_exchange); } @Bean การผูกมัดสาธารณะ DirectBinding1 () {return bindingBuilder.bind (routingQueue1 ()). ถึง (directExchange ()) ด้วย ("ผู้ใช้"); } @Bean การผูกมัดสาธารณะ DirectBinding2 () {return bindingBuilder.bind (routingQueue2 ()). ถึง (directExchange ()) ด้วย ("คำสั่ง"); } // ============== รูปแบบธีม ====================== Public Static Final String Final TOCICATE_QUEUE_1 = "TOBIC_QUEUE_1"; สตริงสุดท้ายคงที่ public public_queue_2 = "topic_queue_2"; สตริงสุดท้ายคงที่ public public_exchange = "topic_exchange"; @Bean สาธารณะ Public Queue TopicQueue1 () {ส่งคืนคิวใหม่ (topic_queue_1, true); } @Bean Public Queue POCITHEQUEUE2 () {ส่งคืนคิวใหม่ (topic_queue_2, true); } @Bean Public TopicexChange TopicexChange () {ส่งคืน topicexChange ใหม่ (TOBIC_EXCHANGE); } @Bean Public Binding TopicBinding1 () {return bindingbuilder.bind (topicqueue1 ()). ถึง (topicexchange ()) ด้วย ("user.add"); } @Bean Public Binding TopicBinding2 () {return bindingbuilder.bind (topicqueue2 ()). ถึง (topicexchange ()) ด้วย ("ผู้ใช้#"); -

RabbitMQ มีโหมดการทำงานหลายโหมดดังนั้นจึงมีการกำหนดค่ามากมาย ผู้อ่านที่ต้องการทราบเกี่ยวกับเนื้อหาที่เกี่ยวข้องสามารถตรวจสอบ "การแนะนำเกี่ยวกับโหมดการทำงานของ RabbitMQ" หรือข้อมูลที่เกี่ยวข้องกับ Baidu ด้วยตนเอง

ผู้ผลิตข้อความ:

 @componentpublic คลาส amqpsender {@autowired ส่วนตัว amqptemplate amqptemplate; / ** * โหมดง่ายส่งข้อความ * * @param ข้อความ */ โมฆะสาธารณะ simplesend (ข้อความสตริง) {this.amqptemplate.convertandsend (amqpconfirguration.simple_queue, ข้อความ); }/** * โหมดเผยแพร่/สมัครสมาชิกส่งข้อความ * * @param ข้อความ */โมฆะสาธารณะ pssend (ข้อความสตริง) {this.amqptemplate.convertandsend (amqpconfirguration.fanout_exchange, "", ข้อความ); } / ** * ส่งในโหมดการกำหนดเส้นทาง * * ข้อความ @param * / โมฆะสาธารณะเส้นทางการกำหนดเส้นทาง (สตริงการกำหนดเส้นทางคีย์, ข้อความสตริง) {this.amqptemplate.convertandsend (amqpconfirguration.direct_exchange, เส้นทาง, ข้อความ); } / ** * ส่งในโหมดธีม * * @param routingKey * @param ข้อความ * / public void topicsEnd (String RoutingKey, ข้อความสตริง) {this.amqptemplate.convertandsend (amqpConfirguration.topic_exchange, RoutingKey, ข้อความ); -

ผู้บริโภคข้อความ:

 @ComponentPublic คลาส amqpreceiver { / ** * การรับโหมดง่าย * * @param message * / @rabbitlistener (queues = amqpconfirguration.simple_queue) โมฆะสาธารณะง่าย ๆ (ข้อความสตริง) {system.out.println }/** * เผยแพร่/สมัครสมาชิกโหมดการรับ * * @param message */@rabbitlistener (queues = amqpconfirguration.ps_queue_1) โมฆะสาธารณะ psreceive1 (ข้อความสตริง) {system.out.println (amqpconfirguration.ps_queue_1 + } @rabbitListener (queues = amqpConfirguration.ps_queue_2) โมฆะสาธารณะ psreceive2 (ข้อความสตริง) {system.out.println (amqpconfirguration.ps_queue_2 + "รับข้อความ:" + ข้อความ); } / ** * การรับโหมดการกำหนดเส้นทาง * * @param message * / @rabbitListener (queues = amqpconfirguration.routing_queue_1) โมฆะสาธารณะเส้นทางการกำหนดเส้นทาง 1 (ข้อความสตริง) {system.out.println (amqpconfirguration.routing_queue_1 + } @rabbitListener (queues = amqpConfirguration.routing_queue_2) โมฆะการกำหนดเส้นทางสาธารณะ 2 (ข้อความสตริง) {system.out.println (amqpconfirguration.routing_queue_2 + "รับข้อความ:" + ข้อความ); } / ** * การรับโหมดหัวข้อ * * ข้อความ @param * / @rabbitlistener (queues = amqpconfirguration.topic_queue_1) โมฆะสาธารณะหัวข้อการรับ 1 (ข้อความสตริง) {system.out.println (amqpconfirguration.topic_queue_1 + } @RabBitListener (queues = amqpConfirguration.topic_queue_2) โมฆะสาธารณะหัวข้อ Receive2 (ข้อความสตริง) {system.out.println (amqpconfirguration.topic_queue_2 + "รับข้อความ:" + ข้อความ); -

ผู้บริโภคข้อความฟังข้อความโดยใช้คำอธิบายประกอบ @RabBitListener

3.4 ทดสอบ

 @runwith (SpringRunner.class) @springboottestpublic คลาส amqptest {@autoWired Private Amqpsender ผู้ส่ง; @Test Public Void TestSimplesEnd () {สำหรับ (int i = 1; i <6; i ++) {this.sender.simplesend ("ทดสอบ simplesend"+i); }} @Test โมฆะสาธารณะ testPSSEND () {สำหรับ (int i = 1; i <6; i ++) {this.sender.pssend ("ทดสอบ pssend"+i); }} @Test โมฆะสาธารณะ testPSSEND () {สำหรับ (int i = 1; i <6; i ++) {this.sender.pssend ("ทดสอบ pssend"+i); }} @Test โมฆะสาธารณะ testRoutingSEND () {สำหรับ (int i = 1; i <6; i ++) {this.sender.routingsEnd ("คำสั่งซื้อ", "การทดสอบเส้นทางการทดสอบ"+i); }} @Test โมฆะสาธารณะ testTopicsEnd () {สำหรับ (int i = 1; i <6; i ++) {this.sender.topicsend ("user.add", "test topicsend"+i); -

ผลการทดสอบถูกข้าม - -

การแจ้งเตือน 1: การเข้าสู่ระบบ access_refused ถูกปฏิเสธโดยใช้กลไกการตรวจสอบความถูกต้องธรรมดา

สารละลาย:

1) โปรดตรวจสอบให้แน่ใจว่าชื่อผู้ใช้และรหัสผ่านถูกต้อง เป็นสิ่งสำคัญที่จะต้องทราบว่าค่าของชื่อผู้ใช้และรหัสผ่านมีช่องว่างหรือแท็บ (ผู้เขียนทดสอบเพราะรหัสผ่านมีอักขระแท็บอีกหนึ่งตัวซึ่งทำให้เกิดความล้มเหลวของการตรวจสอบความถูกต้อง)

2) หากบัญชีทดสอบใช้แขกคุณจะต้องแก้ไขไฟล์ rabbitmq.conf เพิ่มการกำหนดค่า "loopback_users = none none" ในไฟล์

ไม่สามารถเตรียมคิวสำหรับผู้ฟังได้ ไม่ว่าคิวจะไม่มีอยู่หรือนายหน้าจะไม่อนุญาตให้เราใช้งาน

สารละลาย:

เราสามารถเข้าสู่ระบบการจัดการ RabbitMQ และเพิ่มคิวที่เกี่ยวข้องด้วยตนเองในตัวเลือกคิว

ข้างต้นเป็นเนื้อหาทั้งหมดของบทความนี้ ฉันหวังว่ามันจะเป็นประโยชน์ต่อการเรียนรู้ของทุกคนและฉันหวังว่าทุกคนจะสนับสนุน wulin.com มากขึ้น