鏂伴矞鍑虹倝锛岃繖鏄叏缃戣鐨勬渶璇︾粏鐨剆pringboot鏁村悎娑堟伅鏈嶅姟浜嗗惂锛屽缓璁敹钘忥紒

Posted 2020-12-17

鏍囩锛?a href='http://www.mamicode.com/so/1/fanout' title='fanout'>fanout   broker   閲嶅惎   down   鏀寔   private   绛栫暐   void   鏈嶅姟   

springboot鏁村悎activeMq

ActiveMq鏄疉pache鎻愪緵鐨勫紑婧愭秷鎭郴缁熼噰鐢╦ava瀹炵幇锛?/p>

寰堝ソ鍦版敮鎸丣MS锛圝ava Message Service,鍗矹ava娑堟伅鏈嶅姟锛?瑙勮寖

ActiveMq瀹夎锛?a href="http://activemq.apache.org/components/classic/download/" rel="nofollow">http://activemq.apache.org/components/classic/download/ 鍦ㄥ畼缃戜笅杞藉畨瑁呭搴旂殑鐗堟湰

涓嬭浇瀹屾垚鍚庤В鍘嬪氨鍙互浣跨敤

ActiveMq榛樿鐨勭鍙ｅ彿鏄?161锛岀敤鎴峰悕鍜屽瘑鐮侀兘鏄痑dmin 鍦ㄦ湰鏈哄彲浠ヤ娇鐢╤ttp://localhost:8161 鍘昏闂?/p>

springboot鏁村悎ActiveMq

1銆佸鍏ヤ緷璧?/h4>

<dependency>
 ?  <groupId>org.springframework.boot</groupId>
 ?  <artifactId>spring-boot-starter-activemq</artifactId>
</dependency>

2銆佸湪properties鏂囦欢涓厤缃產ctiveMq

spring.activemq.broker-url=tcp://localhost:61616
#濡傛灉鏄偣瀵圭偣锛坬ueue锛夛紝閭ｄ箞姝ゅ榛樿搴旇鏄痜alse锛屽鏋滃彂甯冭闃咃紝閭ｄ箞涓€瀹氳缃负true
spring.activemq.packages.trust-all=true
spring.activemq.user=admin
spring.activemq.password=admin

3銆佺紪鍐檘ueue锛堥槦鍒楋級

@Component
public class QueueBean{
 ? ?//鍒涘缓涓€涓槦鍒楀疄渚? ? ?@Bean
 ? ?Queue queue(){
 ? ? ? ?//杩欓噷璁剧疆鐨勬秷鎭槸闃熷垪鐨勫悕绉? ? ? ? ?return new ActiveMQQueue("hello.javaboy");
 ?  }
}

4銆佸垱寤烘秷鎭殑鍙戦€佽€呬互鍙婃秷璐硅€?/h4>

@Component
public class JmsComponent{
 ? ?//springboot鎻愪緵鐨勬秷鎭ā鏉? ? ?@Autowired
 ? ?JmsMessagingTemplate jmsMessagingTemplate;
 ? ?//鑷繁鍒涘缓鐨勯槦鍒楀疄渚? ? ?@Autowired
 ? ?Queue queue;
 ? ?/**
 ? ? * 鍙戦€佹秷鎭? ? ? * @param message
 ? ? */
 ? ?public void send(Message message){
 ? ? ? ?jmsMessagingTemplate.convertAndSend(this.queue,message);
 ?  }
 ? ?/**
 ? ? * 鎺ユ敹娑堟伅
 ? ? * @param message
 ? ? */
 ? ?//琛ㄧず鐩戝惉璇ラ槦鍒楀悕绉板彂鏉ョ殑娑堟伅
 ? ?@JmsListener(destination = "hello.javaboy")
 ? ?public void readMessage(Message message){
 ? ? ? ?System.out.println(message);
 ?  }
?
}

5銆佷笂杩癕essage瀹炰綋绫?/h4>

public class Message implements Serializable {
 ? ?private String content;//娑堟伅涓讳綋
 ? ?private Date sendDate;//娑堟伅鍙戦€佺殑鏃堕棿
 ? ?//鐪佺暐get銆乻et銆乼ostring鏂规硶
}

6銆佽繘琛屾秷鎭殑鍙戦€佷互鍙婃秷璐?/h4>

鍦ㄦ祴璇曠被涓敞鍏msComponent 璋冪敤send()鏂规硶杩涜娑堟伅鐨勮浆鍙?/p>

@SpringBootTest
class ActivemqApplicationTests {
 ? ?@Autowired
 ? ?JmsComponent jmsComponent;
 ? ?@Test
 ? ?void contextLoads() {
 ? ? ? ?Message message = new Message();
 ? ? ? ?message.setContent("hello activeMq");
 ? ? ? ?message.setSendDate(new Date());
 ? ? ? ?jmsComponent.send(message);
 ?  }
}

棣栧厛鍚姩椤圭洰锛屽湪杩愯娴嬭瘯绫昏繘琛屾秷鎭彂閫侊細

鎺у埗鍙颁細鎵撳嵃娑堟伅鍐呭锛?br/>

?

springboot鏁村悎RabbitMQ

rabbitmq瀹夎姣旇緝绻佺悙锛岃繖閲屼娇鐢╠ocker瀹瑰櫒杩涜瀹夎锛宒ocker瀹夎闈炲父鏂逛究锛屼竴鏉″懡浠ゅ叏閮ㄦ悶瀹?/p>

閫氳繃docker瀹夎rabbitmq

-P(澶)琛ㄧず鑷姩鏄犲皠鍒颁富鏈虹鍙?/p>

docker run -d --hostname my-rabbitmq --name some-rabbitmq -P rabbitmq:3-management

棣栧厛瀵煎叆渚濊禆

 <dependency>
 ? ? <groupId>org.springframework.boot</groupId>
 ? ? <artifactId>spring-boot-starter-amqp</artifactId>
 </dependency>

缂栧啓閰嶇疆鏂囦欢锛?/p>

#閰嶇疆rabbitMQ
spring.rabbitmq.host=localhost
spring.rabbitmq.username=guest
spring.rabbitmq.password=guest
spring.rabbitmq.port=32771

?

RabbitMQ 鍥涚浜ゆ崲妯″紡锛?/h2>

鐩磋繛浜ゆ崲鏈猴細Direct exchange

鎵囧舰浜ゆ崲鏈猴細Fanout exchange

涓讳綋浜ゆ崲鏈猴細Topic exchange

棣栭儴浜ゆ崲鏈猴細Headers exchange

涓嬮潰鍒嗗埆浠嬬粛4涓氦鎹㈡ā寮忥細

1銆丏irect exchange

//Direct绛栫暐锛堝彧杞彂缁檙outingKey鐩稿尮閰嶇殑鐢ㄦ埛锛?@Configuration
public class RabbitDirectConfig {
 ? ?public final static String DIRECTNAME = "javaboy-direct";
 ? ?//娑堟伅闃熷垪
 ? ?@Bean
 ? ?Queue queue(){
 ? ? ? ?//name鍊间负闃熷垪鍚嶇О锛宺outingKey浼氫笌浠栬繘琛屽尮閰? ? ? ? ?return new Queue("hello.RabbitMQ");
 ?  }
 ? ?@Bean
 ? ?Queue queue1(){
 ? ? ? ?return new Queue("hello.RabbitMQ1");
 ?  }
 ? ?@Bean
 ? ?DirectExchange directExchange(){
 ? ? ? ?//绗竴涓弬鏁颁负DIRECTNAME銆佺浜屼釜鍙傛暟琛ㄧず閲嶅惎鍚庢槸鍚︽湁鏁堬紝绗笁鍙傛暟琛ㄧず闀挎椂闂存湭浣跨敤鏄惁鍒犻櫎
 ? ? ? ?return new DirectExchange(DIRECTNAME,true,false);
 ?  }
 ? ?@Bean
 ? ?Binding binding(){
 ? ? ? ?//灏嗛槦鍒梣ueue鍜孌irectExchange缁戝畾鍦ㄤ竴璧? ? ? ? ?return BindingBuilder.bind(queue()).to(directExchange()).with("direct");
 ?  }
 ? ?@Bean
 ? ?Binding binding1(){
 ? ? ? ?//灏嗛槦鍒梣ueue鍜孌irectExchange缁戝畾鍦ㄤ竴璧? ? ? ? ?return BindingBuilder.bind(queue1()).to(directExchange()).with("direct");
 ?  }
?
}

2銆侀厤缃秷璐硅€匘irectReceiver锛?/h4>

//閰嶇疆娑堣垂鑰?@Component
public class DirectReceiver {
 ? ?//鍙洃鍚琿ueue()闃熷垪鐨勬秷鎭? ? ?@RabbitListener(queues = "hello.RabbitMQ")
 ? ?public void hanlder(String msg){
 ? ? ? ?System.out.println("hanlder>>>"+msg);
?
 ?  }
 ? ?//鍙洃鍚琿ueue1()闃熷垪鐨勬秷鎭? ? ?@RabbitListener(queues = "hello.RabbitMQ1")
 ? ?public void hanlder1(String msg){
 ? ? ? ?System.out.println("hanlder1>>>"+msg);
?
 ?  }
}

娴嬭瘯浠ｇ爜锛?/p>

鍦╯pringboot鐨勬祴璇曠被涓敞鍏abbitTemplate(springboot鎻愪緵鐨凴abbitMQ妯℃澘)

 @Autowired
 ? ?RabbitTemplate rabbitTemplate;
 ? ?@Test
 ? ?void contextLoads() {
 ? ? ? ?//涓や釜鍙傛暟绗竴涓槸routingKey銆佺浜屼釜涓烘秷鎭唴瀹? ? ? ? ?rabbitTemplate.convertAndSend("hello.RabbitMQ","hello RabbitMQ test");
 ? ? ? ?rabbitTemplate.convertAndSend("hello.RabbitMQ1","hello RabbitMQ test222");
 ?  }

鍚姩椤圭洰鍚庯紝杩愯娴嬭瘯绫诲彲浠ョ湅鍒板彧鏈変笌routingkey鐩稿尮閰嶇殑娑堣垂鑰呭彈鍒颁簡瀵瑰簲鐨勬秷鎭細

?

2銆丗anout exchange

Fanout绛栫暐锛堝彧瑕佹槸涓庝粬缁戝畾鐨勯槦鍒楋紝閮戒細鏀跺埌娑堟伅涓巖outingKey鏃犲叧锛?/h2>

1銆侀厤缃甊abbitFanoutConfig锛?/h4>

//Fanout绛栫暐锛堝彧瑕佹槸涓庝粬缁戝畾鐨勯槦鍒楋紝閮戒細鏀跺埌娑堟伅涓巖outingKey鏃犲叧锛?@Configuration
public class RabbitFanoutConfig {
 ? ?public final static String FANOUTNAME = "javaboy-fanout";
    //閰嶇疆浜嗕袱涓秷鎭槦鍒梣ueueOne鍜宷ueueTwo
 ? ?@Bean
 ? ?Queue queueOne(){
 ? ? ? ?return new Queue("queue-one");
 ?  }
 ? ?@Bean
 ? ?Queue queueTwo(){
 ? ? ? ?return new Queue("queue-two");
 ?  }
 ? ?@Bean
 ? ?FanoutExchange fanoutExchange(){
 ? ? ? ?return new FanoutExchange(FANOUTNAME,true,false);
 ?  }
 ? ?//灏嗕袱涓槦鍒椾笌FanoutExchange缁戝畾
 ? ?@Bean
 ? ?Binding bindingOne(){
 ? ? ? ?return BindingBuilder.bind(queueOne()).to(fanoutExchange());
 ?  }
 ? ?@Bean
 ? ?Binding bindingTwo(){
 ? ? ? ?return BindingBuilder.bind(queueTwo()).to(fanoutExchange());
 ?  }
}

2銆侀厤缃秷璐硅€匜anoutReceiver锛?/h4>

//閰嶇疆娑堣垂鑰?@Component
public class FanoutReceiver {
    //涓や釜娑堣垂鑰呭垎鍒洃鍚袱涓笉鍚岀殑闃熷垪
 ? ?@RabbitListener(queues = "queue-one")
 ? ?public void hanlder1(String msg){
 ? ? ? ?System.out.println("FanoutReceiver:hanlder1>>>"+msg);
?
 ?  }
 ? ?@RabbitListener(queues = "queue-two")
 ? ?public void hanlder2(String msg){
 ? ? ? ?System.out.println("FanoutReceiver:hanlder2>>>"+msg);
?
 ?  }
}

3銆佹祴璇曠被锛?/h4>

@Test
 ? ?void rabbitFanout(){
 ? ? ? ?//涓変釜鍙傛暟琛ㄧずRabbitFanoutConfig鐨勫悕绉般€乺outingkey銆佹秷鎭唴瀹? ? ? ? ?rabbitTemplate.convertAndSend(RabbitFanoutConfig.FANOUTNAME,null,"hello fanout test");
 ?  }

璇ユ柟寮忎笌routingkey鏃犲叧鎵€鏈夊啓null鍗冲彲

鏌ョ湅杈撳嚭锛氬彲浠ョ湅鍒颁袱涓秷璐硅€呴兘鏀跺埌浜嗘秷鎭?/p>

?

3銆乀opic exchange

topic绛栫暐鍙互鏍规嵁routingKey鐨勮鍒欙紙閫氶厤绗︽柟寮忥級杩涜鍘诲尮閰嶉槦鍒楄繘琛岃浆鍙戣鍒欎负.#. *涓哄崟璇嶏紝#琛ㄧず妯＄硦鍖归厤

渚嬪routingkey涓猴細xiaomi.# 閭ｄ箞甯︽湁xiaomi.寮€澶寸殑闃熷垪閮戒細鏀跺埌璇ユ秷鎭?/p>

routingkey涓猴細#.phone.# 琛ㄧず娑堟伅鐨剅outingKey涓甫鏈塸hone鏃?灏变細鍘诲尮閰嶅甫鏈塸hone鐨勯槦鍒?/p>

1銆侀厤缃甊abbitTopicConfig锛?/h4>

/topic绛栫暐鍙互鏍规嵁routingKey鐨勮鍒欙紙閫氶厤绗︽柟寮忥級杩涜鍘诲尮閰嶉槦鍒楄繘琛岃浆鍙戣鍒欎负*.#.*
 ? ?//*涓哄崟璇嶏紝#琛ㄧず妯＄硦鍖归厤
@Configuration
public class RabbitTopicConfig {
 ? ?public final static String TOPICNAME = "javaboy-topic";
?
 ? ?@Bean
 ? ?TopicExchange topicExchange(){
 ? ? ? ?return new TopicExchange(TOPICNAME,true,false);
 ?  }
 ? ?@Bean
 ? ?Queue xiaomi(){
 ? ? ? ?return new Queue("xiaomi");
 ?  }
 ? ?@Bean
 ? ?Queue huawei(){
 ? ? ? ?return new Queue("huawei");
 ?  }
 ? ?@Bean
 ? ?Queue phone(){
 ? ? ? ?return new Queue("phone");
 ?  }
?
 ? ?@Bean
 ? ?Binding xiaomiBinding(){
 ? ? ? ?//xiaomi.#锛氳〃绀烘秷鎭殑routingKey鏄互xiaomi寮€澶寸殑灏变細璺敱鍒皒iaomi鐨勯槦鍒? ? ? ? ?return BindingBuilder.bind(xiaomi()).to(topicExchange()).with("xiaomi.#");
 ?  }
 ? ?@Bean
 ? ?Binding huaweiBinding(){
 ? ? ? ?return BindingBuilder.bind(huawei()).to(topicExchange()).with("huawei.#");
 ?  }
 ? ?@Bean
 ? ?Binding phoneBinding(){
 ? ? ? ?//#.phone.#锛氳〃绀烘秷鎭殑routingKey涓甫phone鐨勯兘浼氳矾鐢卞埌phone鐨勯槦鍒? ? ? ? ?return BindingBuilder.bind(phone()).to(topicExchange()).with("#.phone.#");
 ?  }
}

2銆侀厤缃秷璐硅€匱opicReceiver锛?/h4>

@Component
public class TopicReceiver {
 ? ?//鍒嗗埆鐩戝惉鍚嶇О涓簒iaomi銆乭uawei銆乸hone鐨勯槦鍒? ? ?@RabbitListener(queues = "xiaomi")
 ? ?public void handlerXM(String msg){
 ? ? ? ?System.out.println("TopicReceiver:handlerXM>>>"+msg);
 ?  }
 ? ?@RabbitListener(queues = "huawei")
 ? ?public void handlerHW(String msg){
 ? ? ? ?System.out.println("TopicReceiver:handlerHW>>>"+msg);
 ?  }
 ? ?@RabbitListener(queues = "phone")
 ? ?public void handlerPHONE(String msg){
 ? ? ? ?System.out.println("TopicReceiver:handlerPHONE>>>"+msg);
 ?  }
}

3銆佹祴璇曠被锛?/h3>

@Test
 ? ?void rabbitTopic(){
 ? ? ? ?//鏍规嵁鍖归厤瑙勫垯璇ユ秷鎭彧鑳借xiaomi鐨勯槦鍒楁敹鍒? ? ? ? ?rabbitTemplate.convertAndSend(RabbitTopicConfig.TOPICNAME,"xiaomi.news","灏忕背鏂伴椈");
 ? ? ? ?//鏍规嵁鍖归厤瑙勫垯璇ユ秷鎭彧鑳借phone鐨勯槦鍒楁敹鍒? ? ? ? ?rabbitTemplate.convertAndSend(RabbitTopicConfig.TOPICNAME,"vivo.phone","vivo鎵嬫満");
 ? ? ? ?//鏍规嵁鍖归厤瑙勫垯璇ユ秷鎭彲浠ュ埆huawei鍜宲hone涓や釜闃熷垪鏀跺埌
 ? ? ? ?rabbitTemplate.convertAndSend(RabbitTopicConfig.TOPICNAME,"huawei.phone","鍗庝负鎵嬫満");
?
 ?  }

鏌ョ湅杈撳嚭锛?/p>

?

鍙互鐪嬪埌routingkey涓篽uawei.phone鐨勬秷鎭尮閰嶄簡涓や釜闃熷垪锛屽叾浠栦袱涓兘鍙尮閰嶄簡涓€涓槦鍒?/p>

4銆丠eaders exchange

璇ユā寮忔槸鏍规嵁璺敱瑙勫垯鐨刪eader杩涜鍖归厤鐨勶紝鍦ㄨ繘琛屽尮閰嶇殑鏃跺€欓渶瑕佷紶鍏ヤ竴涓猰ap闆嗗悎锛宺outingkey鍘诲尮閰峬ap鍗冲彲涓殑key value锛屽尮閰嶈鍒欏彲浠ヤ娇any鎴栬€卆ll锛宎ny琛ㄧず鍙鍖呭惈浠绘剰淇℃伅灏卞彲浠ワ紝all琛ㄧず鎵€鏈変俊鎭兘蹇呴』鍖归厤

1銆侀厤缃甊abbitHeaderConfig锛?/h4>

@Configuration
public class RabbitHeaderConfig {
 ? ?public final static String HEADERNAME = "javaboy-header";
?
 ? ?@Bean
 ? ?HeadersExchange headersExchange(){
 ? ? ? ?return new HeadersExchange(HEADERNAME,true,false);
 ?  }
 ? ?//鍒嗗埆鍒涘缓涓や釜涓嶅悓header鐨勯槦鍒? ? ?@Bean
 ? ?Queue queueName(){
 ? ? ? ?return new Queue("name-queue");
 ?  }
 ? ?@Bean
 ? ?Queue queueAge(){
 ? ? ? ?return new Queue("age-queue");
 ?  }
 ? ?@Bean
 ? ?Binding bindingName(){
 ? ? ? ?Map<String,Object> map = new HashMap<>();
 ? ? ? ?map.put("name","hello");
 ? ? ? ?//琛ㄧず濡傛灉routingKey鍖归厤鐨刴ap闆嗗悎涓殑key value 灏变細灏嗘秷鎭浆鍙戝埌瀵瑰簲鐨勮矾鐢变笂
 ? ? ? ?return BindingBuilder.bind(queueName()).to(headersExchange()).whereAny(map).match();
 ?  }
?
 ? ?@Bean
 ? ?Binding bindingAge(){
 ? ? ? ?return BindingBuilder.bind(queueAge()).to(headersExchange()).where("age").exists();
 ?  }
}

2銆佸垱寤烘秷璐硅€匟eaderReceiver:

@Component
public class HeaderReceiver {
 ? ?@RabbitListener(queues = "name-queue")
 ? ?public void handlerName(byte[] msg){
 ? ? ? ?System.out.println("HeaderReceiver:handlerName>>>>"+new String(msg,0,msg.length));
 ?  }
 ? ?@RabbitListener(queues = "age-queue")
 ? ?public void handlerAge(byte[] msg){
 ? ? ? ?System.out.println("HeaderReceiver:handlerAge>>>>"+new String(msg,0,msg.length));
 ?  }
}

3銆佹祴璇曚唬鐮侊細

@Test
 ? ?public void rabbitHeader(){
 ? ? ? ?//璁剧疆娑堟伅锛屽苟涓旇缃甴eader锛宻etHeader("name","hello")鍒嗗埆琛ㄧずmap闆嗗悎涓殑key銆乿alue
 ? ? ? ?Message nameMessage = 
 ? ? ? ? ? ?MessageBuilder.withBody("hello name".getBytes()).setHeader("name","hello").build();
 ? ? ? ?Message ageMessage =
 ? ? ? ? ? ?MessageBuilder.withBody("hello 99 age".getBytes()).setHeader("age","99").build();
 ? ? ? ?rabbitTemplate.send(RabbitHeaderConfig.HEADERNAME,null,nameMessage);
 ? ? ? ?rabbitTemplate.send(RabbitHeaderConfig.HEADERNAME,null,ageMessage);
 ?  }

鏌ョ湅杈撳嚭锛?/p>

?

鏀瑰彉setheader涓殑鍊兼煡鐪嬬粨鏋滐細

 Message nameMessage = 
 ? ? ? ? ?  MessageBuilder.withBody("hello name".getBytes()).setHeader("name","javaboy").build();

?

鍙互鐪嬪埌鍥犱负key銆乿alue鍖归厤涓嶄笂鍙墦鍗颁簡涓€鏉℃秷鎭€?br/>

鏈€鍚?/h2>

澶у鐪嬪畬鏈変粈涔堜笉鎳傜殑鍙互鍦ㄤ笅鏂圭暀瑷€璁ㄨ锛屼篃鍙互鍏虫敞鎴戠淇￠棶鎴戯紝鎴戠湅鍒板悗閮戒細鍥炵瓟鐨勩€備篃娆㈣繋澶у鍏虫敞鎴戠殑鍏紬鍙凤細鍓嶇▼鏈夊厜锛岄噾涓夐摱鍥涜烦妲介潰璇曞锛屾暣鐞嗕簡1000澶氶亾灏嗚繎500澶氶〉pdf鏂囨。鐨凧ava闈㈣瘯棰樿祫鏂欙紝鏂囩珷閮戒細鍦ㄩ噷闈㈡洿鏂帮紝鏁寸悊鐨勮祫鏂欎篃浼氭斁鍦ㄩ噷闈€傝阿璋綘鐨勮鐪嬶紝瑙夊緱鏂囩珷瀵逛綘鏈夊府鍔╃殑璇濊寰楀叧娉ㄦ垜鐐逛釜璧炴敮鎸佷竴涓嬶紒