SpringBoot는 RabbitMQ 샘플 코드를 통합합니다

개요

1. RabbitMQ는 일반 프로토콜을 통해 완전히 다른 응용 프로그램간에 데이터를 공유하는 데 사용되는 오픈 소스 메시지 브로커 및 큐 서버입니다.
2. AMQP 프로토콜을 현실적으로 구현하고 Mozilla Public License 오픈 소스 프로토콜을 따르며 여러 언어를 지원하고 Spring과 쉽게 통합 될 수 있습니다.
3. 메시지 큐는 RabbitMQ와 같은 메시지 프록시 서버를 통해 응용 프로그램간에 라우팅되는 메시지를 사용하여 응용 프로그램을 연결합니다.

기본 개념

브로커

메시지 큐 서버 엔티티는 데이터를 처리하는 데 사용됩니다

vhost

RabbitMQ 서버에서 만든 가상 메시지 호스트에는 자체 권한 메커니즘이 있습니다. 여러 사용자의 허가를 위해 브로커에서 여러 Vhost를 열 수 있으며 Vhosts도 완전히 격리됩니다.

생산자

메시지 통신을위한 데이터를 생성합니다

채널

메시지 채널, 여러 채널을 AMQP로 설정할 수 있으며 각 채널은 세션 작업을 나타냅니다.

교환

직접

라우팅 키로 지정된 대기열로 메시지를 전달합니다

팬 아웃

팬 아웃

모든 바인딩 대기열에 메시지를 전달하는 것은 방송 방식과 유사합니다.

주제

주제

규칙에 따라 메시지 전달. 이 규칙은 대부분 패턴 일치하며 더 유연하게 보입니다.

대기줄

대기줄

1. 대기열은 RabbitMQ의 내부 대상으로 메시지를 저장합니다.
2. 소비자가 동적으로 증가함에 따라 대기열은 라운드 로빈 방식으로 여러 소비자에게 수신 된 메시지를 고르게 할당합니다.

제본

스위치와 큐의 관계를 나타냅니다. 바인딩시, 라우팅 키와 일치하는 추가 파라미터 바인딩 키가 제공됩니다.

소비자

메시지 데이터를 읽으려면 메시지 대기열을 듣습니다.

SpringBoot에서 3 가지 교환 모드 (Fanout, Direct, Topic) 구현

pom.xml의 스프링 부트 스타터 -amqp 참조

 <pectionency> <groupId> org.springframework.boot </groupid> <artifactid> 스프링 부트 스타터-amqp </artifactid> </fectionency>

RabbitMQ 구성을 추가하십시오

 봄 : RabbitMQ : 호스트 : LocalHost 포트 : 5672 사용자 이름 : 게스트 비밀번호 : 게스트

직접

직접 모드에서는 일반적으로 정의하기 위해 큐 만 필요합니다. 스위치를 바인딩하지 않고 내장 스위치 (defaultexchange)를 사용하십시오.

 @ConfigurationPublic Class Rabbitp2pconfigure {public static final String queue_name = "p2p-queue"; @bean public queue queue () {return new queue (queue_name, true); }}

 @runwith (springrunner.class) @springboottest (classs = bootcoretestApplication.class)@slf4jpublic class RabbitTest {@autowired private amqptemplate amqptemplate; / *** send*/ @test public void sendlazy ()가 중단 된 결과 {City City = New City (234556666L, "Direct_name", "Direct_code"); amqptemplate.convertandSend (RabbitlazyConfigure.queue_name, City); } / *** 수신* / @test public void leceper () throws InterruptedException {object obj = amqptemplate.receiveAndConvert (rabbitlazyConfigure.queue_name); assert.notnull (obj, ""); log.debug (obj.tostring ()); }}

적용 가능한 시나리오 : 지점 간 포인트

팬 아웃

Fanout 모드는 여러 큐를 동일한 스위치에 바인딩해야합니다.

 @ConfigurationPublic 클래스 RabbitFanoutConfigure {public static final String exchange_name = "Fanout-Exchange"; 공개 정적 최종 문자열 fanout_a = "fanout.a"; 공개 정적 최종 문자열 fanout_b = "fanout.b"; 공개 정적 최종 문자열 fanout_c = "fanout.c"; @bean public equeue amessage () {return new queue (fanout_a); } @bean public queue bmessage () {return new queue (fanout_b); } @bean public queue cmessage () {return new queue (fanout_c); } @Bean 공개 fanOutExChange fanOutExChange () {return new fanOutexChange (exchange_name); } @bean public binding bindingexchangea (Queue Amessage, FanOutexChange fanOutexChange) {return bindingBuilder.bind (Amessage) .to (fanOutexChange); } @bean public binding bindingexchangeb (Queue bmessage, fanOutexChange fanOutexChange) {return bindingBuilder.bind (bmessage) .to (fanOutexChange); } @bean public binding bindingexchangec (Queue cmessage, fanOutexChange fanOutexChange) {return bindingBuilder.bind (cmessage) .to (fanOutexChange); }}

보내는 사람

 @slf4jpublic 클래스 발신자 {@autowired private amqptemplate rabbittemplate; public void sendfanout (객체 메시지) {log.debug ( "팬 아웃 보내기 시작 <" + message + ">"); Rabbittemplate.convertandSend (RabbitFanoutConfigure.exchange_name, "", 메시지); }}

@RabbitListener를 사용하여 소비 할 여러 대기열을들을 수 있습니다.

 @slf4j @rabbitlistener (queues = {rabbitfanoutconfigure.fanout_a, rabbitfanoutconfigure.fanout_b, rabbitfanoutconfigure.fanout_c}) public class receiver {@rabbithandler public void invidessage (문자열 메시지) { ""< "" + ">"); }}

해당 시나리오
-대규모 멀티 사용자 온라인 (MMO) 게임은이를 사용하여 업데이트 순위와 같은 글로벌 이벤트를 처리 할 수 있습니다.
- 스포츠 뉴스 웹 사이트는이를 사용하여 거의 실시간으로 모바일 클라이언트에 점수 업데이트를 배포 할 수 있습니다.
- 배포 시스템은이를 사용하여 다양한 상태 및 구성 업데이트를 방송합니다.
- 그룹 채팅 중에는 그룹 채팅에 참여하는 사용자에게 메시지를 배포하는 데 사용됩니다.

주제

이 패턴은 비교적 복잡합니다. 간단히 말해서, 각 대기열에는 고유 한 주제가 있습니다. 모든 메시지에는 "제목"이 있습니다. Exchange는 메시지를 Fuzzy Matches Routekey와 관련된 주제의 대기열로 전달합니다.

바인딩 할 때 "주제.# ("# "는 0 또는 여러 키워드를 의미하고"*"는 키워드를 의미합니다.)

 @ConfigurationPublic 클래스 RabbitTopicConfigure {public static final String exchange_name = "Topic-Exchange"; 공개 정적 최종 문자열 주제 = "주제"; public static final string topic_a = "topic.a"; public static final string topic_b = "topic.b"; @bean public equeue queuetopic () {return new queue (rabbittopicconfigure.topic); } @bean public equeue queuetopica () {return new queue (rabbittopicconfigure.topic_a); } @bean public equeue queuetopicb () {return new queue (rabbittopicconfigure.topic_b); } @bean public topicexchange exchange () {topicexchange topicexchange = new topicexchange (Exchange_name); topicexchange.setDelayed (true); 새로운 topicexchange (Exchange_name)를 반환합니다. } @bean public binding bindingexchangetopic (대기열 queuetopic, topicexchange exchange) {return bindingbuilder.bind (queuetopic) .to (exchange) .with (rabbittopicconfigure.topic); } @bean public binding bindingexchangetopics (대기열 queuetopica, topicexchange exchange) {return bindingbuilder.bind (queuetopica) .to (exchange) .with ( "topic.#"); }}

동시에 세 큐를 들어보십시오

 @slf4j @rabbitlistener (queues = {rabbittopicconfigure.topic, rabbittopicconfigure.topic_a, rabbittopicconfigure.topic_b}) 공개 클래스 수신기 {@RabBithandler public void 미주당 (문자열 메시지) {log.debug ( " + 메시지 +"> "); }}

테스트를 통해 우리는 찾을 수 있습니다

 @runwith (springrunner.class) @springboottest (classes = bootcoretestApplication.class) 공개 클래스 RabbitTest {@autowired private amqptemplate Rabbittemplate; @test public void sendall () {rabbittemplate.convertandsend (rabbittopicconfigure.exchange_name, "topic.test", "모두 보내기"); } @test public void sendtopic () {rabbittemplate.convertandSend (rabbittopicconfigure.exchange_name, rabbittopicconfigure.topic, "send topic"); } @Test public void sendTopica () {Rabbittemplate.convertandSend (RabbittopicConfigure.exchange_name, rabbittopicconfigure.topic_a, "topica"); }}

해당 시나리오
- 판매 지점과 같은 특정 지리적 위치에 대한 데이터 배포
- 여러 근로자가 완료 한 무대 뒤 작업, 특정 특정 작업을 처리하는 각 근로자
- 주가 업데이트 (및 기타 유형의 재무 데이터 업데이트)
- 카테고리 또는 태그와 관련된 뉴스 업데이트 (예 : 특정 스포츠 또는 팀의 경우)
- 클라우드에서 다양한 유형의 서비스 조정
- 각 빌더가 하나의 특정 아키텍처 또는 시스템 만 처리 할 수있는 분산 아키텍처/시스템 기반 소프트웨어 패키지.

지연 대기열

소비 지연 :

1. 사용자가 주문을 생성하는 경우 주문의 지불 상태를 확인하는 데 시간이 걸립니다. 주문이 아직 지불되지 않으면 주문을 제 시간에 닫아야합니다.
2. 사용자 등록이 성공하면 사용자의 사용을 확인하는 데 일주일이 걸립니다. 사용자의 활동이 낮은 것으로 밝혀지면 이메일이나 문자 메시지를 보내서 사용자가 사용하도록 상기시킵니다.

지연된 재 시도 :

1. 소비자가 큐에서 메시지를 소비하지 못하지만 일정 기간 동안 지연 후 자동으로 다시 시도하려는 경우.
2. 지연 대기열을 사용하지 않으면 폴링 스캐너를 통해서만 수행 할 수 있습니다. 이 솔루션은 개발자가 사용할 수있는 통합 서비스를 만드는 데 우아하거나 편리하지 않습니다. 그러나 지연 대기열로 쉽게 할 수 있습니다.

스위치 지연 속성을 true로 설정하십시오

 @ConfigurationPublic 클래스 RabbitlazyConfigure {public static final String queue_name = "Lazy-Queue-t"; 공개 정적 최종 문자열 Exchange_name = "Lazy-Exchange-T"; @bean public queue queue () {return new queue (queue_name, true); } @bean public directexChange defaultexchange () {DirectExchange DirectExchange = new DirectExchange (Exchange_name, true, false); DirectExchange.setDelayed (true); DirectExChange를 반환합니다. } @bean public binding binding () {return bindingBuilder.bind (queue ()). to (defaultexchange ()). }}

보낼 때 지연 시간을 설정하십시오

 @slf4jpublic 클래스 발신자 {@autowired private amqptemplate rabbittemplate; public void sendlazy (object msg) {log.debug ( "게으른 메시지 보내기 시작 <" + msg + ">"); Rabbittemplate.convertandSend (RabbitlazyConfigure.exchange_name, RabbitlazyConfigure.queue_name, msg -> {message.getMessageProperties (). SetHeader ( "x -delay", 10000); Return Message;}); }}

마치다

다양한 사용 사례에 대해 공식 문서를 직접 확인하십시오.

위는이 기사의 모든 내용입니다. 모든 사람의 학습에 도움이되기를 바랍니다. 모든 사람이 wulin.com을 더 지원하기를 바랍니다.