全新java高級(jí)架構(gòu)師項(xiàng)目實(shí)戰(zhàn)智能運(yùn)營(yíng)助手rocketmq基礎(chǔ)

每特學(xué)院Java互聯(lián)網(wǎng)架構(gòu)培 老MQ概序提供豐富的消息拉取模式高效的訂閱者水平擴(kuò)展能力實(shí)時(shí)的消息訂閱機(jī)制億級(jí)消息MQ包含的組NameNameServerNameServershare-nothingNameServerNameServerNameServer所有狀態(tài)都從Broker上報(bào)而來，本身 NameServerBrokerBrokergroupgroupmaster，slave。masterslaveslavemastermaster客戶端消費(fèi)可以從master和slave消費(fèi)。在默認(rèn)情況下，消費(fèi)者都從master消費(fèi)，在master掛后，客戶端由于從NameServer中感知到Broker，就會(huì)從slave消費(fèi)。Broker向所有的NameServer結(jié)點(diǎn)建立長(zhǎng)連接，Topic信息。MQ優(yōu)MQ環(huán)境安Hostviservicenetwork注意Error:NosuitabledevicefoundnodevicefoundforconnectionSystemifconfigaifconfigaMACHWADDRvim/etc/sysconfig/network-scripts/ifcfg-eth0ifconfigMACHWADDR值；#上傳alibaba-mq-3.2.6.tar.gz文件至/usr/local#tar-zxvfalibaba-mq-3.2.6.tar.gz-C/usr/local#mvalibaba-mqalibaba-mq-3.2.6#ln-salibaba-mq-3.2.6mq創(chuàng)建路徑【兩臺(tái)機(jī)器#mkdir #mkdir/usr/local/mq/store/consumequeue#mkdir/usr/local/mq/store/indexMQ#vim/usr/local/mq/conf/2m-noslave/perties#mkdir-p/usr/local/#cd mq/conf&&sed-i mq#g'XX:PermSize=128m-XX:MaxPermSize=320m"vim/usr/local/XX:PermSize=128m-XX:MaxPermSize=320m"#cd/usr/local/mq/bin#nohupshmqnamesrv&cd/usr/local/#nohupshmqbroker-c #netstat-ntlp#jps#tail-f-n500/usr/local/mq/logs/#tail-f-n500/usr/local/mq/logs/cd/usr/local/#nohupshmqbroker-c #netstat-ntlp#jps#tail-f-n500/usr/local/mq/logs/mqlogs/broker.log#tail-f-n500/usr/local/ MQ mq-web-console部署到 jdkviexportexportexportPATH=$JAVA_HOME/bin:$PATHsourceJava操 mq-<artifactId>logback-<artifactId>logback-publicpublicclassProducerpublicstaticvoidmain(String[]args)throwsExceptionDefaultMQProducerproducer=newDefaultMQProducer("rmq-tryfor(inti=0;i<10;i++){Thread.sleep(1000);每秒發(fā)送一次MQMessagemsgnewMessage("itmayiedu-topic", 名"TagA",tag臨時(shí))//SendResultsendResult=producer.send(msg);}}}catch(Exceptione){}}}publicpublicclassConsumerpublicstaticvoidmain(String[]args)throwsExceptionDefaultMQPushConsumerconsumer=newDefaultMQPushConsumer("rmq-consumer.subscribe("itmayiedu-topic","TagA");consumer.registerMessageListener(newMessageListenerConcurrently()publicConsumeConcurrentlyStatusconsumeMessage(List<MessageExt>msgs,ConsumeConcurrentlyContextcontext){for(MessageExtmsg:msgs)System.out.println(msg.getMsgId()+"---"+new}return}System.out.println("ConsumerStarted.");}}MQ重試機(jī)MQMQpublicpublicclassConsumerpublicstaticvoidmain(String[]args)throwsExceptionDefaultMQPushConsumerconsumer=newDefaultMQPushConsumer("rmq-consumer.subscribe("itmayiedu-topic","TagA");consumer.registerMessageListener(newMessageListenerConcurrently()publicConsumeConcurrentlyStatusconsumeMessage(List<MessageExt>msgs,ConsumeConcurrentlyContextcontext){for(MessageExtmsg:msgs)System.out.println(msg.getMsgId()+"---"+new}tryinti=1/}catch(Exceptione){需要重return}不需要重return}System.out.println("ConsumerStarted.");}}IDIDMQIDIDIDID、訂單號(hào)msg.setKeys生產(chǎn)者publicpublicclassProducerpublicstaticvoidmain(String[]args)throwsExceptionDefaultMQProducerproducer=newDefaultMQProducer("rmq-producersetNamesrvAddr("192168110195:9876;192168110199:9876");producersetInstanceName("producer");producerstart();try{for(inti=0;i<1;i++)Threadsleep(1000秒發(fā)送一次MessagemsgnewMessage("itmayiedu-topic 名"TagAtag時(shí)("itmayiedu-6igetBytes(body內(nèi)msgsetKeys(SystemcurrentTimeMillis()+"");SendResultsendResult=producersend(msg);Systemoutprintln(sendResulttoString());}}catch(Exceptione)e}producer}}消費(fèi)者staticstaticprivateMap<String,String>logMap=newpublicstaticvoidmain(String[]args)throwsExceptionDefaultMQPushConsumerDefaultMQPushConsumerconsumer=newDefaultMQPushConsumer("rmq-consumer.subscribe("itmayiedu-topic","TagA");consumer.registerMessageListener(newMessageListenerConcurrently(){publicConsumeConcurrentlyStatusconsumeMessage(List<MessageExt>msgs,ConsumeConcurrentlyContextcontext){Stringkey=null;StringmsgId=null;try{for(MessageExtmsg:msgs){key=msg.getKeys();if(logMap.containsKey(key))System.out.println(