RabbitMq开发入门资料

目录

1资产管理模块JavaBean..................................2

2Mq生产者和消费者........................................4

2.1自定义发送类................................................................4

2.2send类(producer)........................................................................................................................5

2.3监听类(consumer)........................................................................................................................6

3Mq的配置文件详解........................................7

3.1连接配置...................................................................7

3.2重试策略和消息转换器.......................................................8

3.3定义queue和交换机.........................................................8

3.4监听器注册和队列配对.......................................................9

4初始化开发环境(内网)....................................9

4.1Git拉取远程仓库项目........................................................9

4.2如何修改项目名............................................................10

4.3Maven仓库的修改..........................................................12

4.4注意事项..................................................................12

1资产管理模块JavaBean

熟悉hap开发或者ssm开发的，会很熟悉，根据模块来建包，例如资产管理模块ast,其

中dto、service%mapper、controllers是典型的mvc开发模式，方便后期维护。

▼・java

▼EScom.hand.his

ito根据Mybatis对象关系映射的规则编写dto类

▼Blast

Elcontrollersmapper一般维承Mapper〈DTO＞即可

▼Eldto

QPurContractStoreHeadservice业务接口由于继承工BaseService故可以使

QPurContractStoreLlne用框架自带的增删改查也行

▼EHmapper

OPurContractStoreHeadMapper

❶PurContractStoreLineMapper

▼Diservice

▼aimp!

❻PurContractStoreHeadServicelmpI

0PurContractStoreLineServicelmpI

❶IPurContractStoreHeadService

❶IPurContractStoreLineService

2发送原理:

MQ的发送消息的方式这里主要介绍topic模式

Q1

P代表消息发送者，x代表交换机，Q1和Q2代表队列，C1和C2代表消息的接收者。

消息的发送经历三个步骤：

第一步是发送消息时指定接收消息的交换机exchange;

2

JSONObj«ctj*on-MVJS0N0bj»ctO：

j»<».put("data*,si)：_

JSOCLput("isXeedResponse*.true)]

json.put(*reque*tldv,1001):

String*3-json.toJSGKStrincO;

Sytte*.out.println(r<..................1111;

Systra.out.println(13):

SrstM.out...............■—》.)；

〃・码〔

aASE64Encod«rba»«MEncod«r-neBASEMEncoderO;

Stringencode-ba»«MEncod»r.«0(0<1«(*3.getBytesCcharwtNamr

»end«r<o«pofMnta.stndMess«<«(EXCHANGE,ROUTER,encod*);

catch(Exceptione)(

假设图中的encode就是我们要发送的消息，需要传入的前两个参数分别是交换机名和

routingkey,而它调用的sendMessage方法可如下图：

■port©rt.sprintfreaevork.MQP.core.AaQpTrapHte:

or，.eprincfraBmrk.b««nt.factory,annotation.Autovired;

rtorg.«prin«frMWvor*i.ttereotyp*.Co«pon«nt;

cUttM*a>M«S«nd«rCoapon«nt•(

privateAaqpT«apl«t<••qpTtipUtt;

publicvoi4••odtttiMfe(Stringexchann.Stringrwittr,Stri»g•»»«•)(

■Mu>T«w)let«.convertAnd$«nd(«xchann<rout”,■••••<•>；

于此同时队列已经和交换机绑定了对应的routingkey,于是第二步就是交换机通过

routingkey找到接收消息的queue,也就是找到队列，从而向其发送消息；

第三部是接收者收消息，要收消息接收者只要指定监听的哪一个队列，就可以从哪一个队

列中获取消息。

2.1Topic模式

在rabbit:bindings标签里pattern代表routingkey,在这里也可以用通配符*和#,*

代表一个，而#号则可以代表多个，例如pattern="*.test”表示可以匹配交换机名为

“1.test”和"2.test”，“3.test”等等的，而pattern="也test”则可以匹配如"1.test”，

3

“21.lesl”，“1322.lesl”等等之类的。只要在这里绑定了之后发数据其实就是

直接调用.AmqpTemplate类下的convertAndSend(exchange,router,message)方法而己，

这里的变量，前两个要指定为配置文件里的routingkey和队列名，后面的就是要发送的消

息。

3Mq生产者和消费者

Mq发送方即producer,接收方•(监听方)即consumer,可以按照功能模块进行分模块处理,

也可以，下面有三个包,分别为components、controllers、listnser,即消息的发送模版类、

发送、接收，可以自定义包名，一般发送和接收要分开来写，符合开发规范。

■1java

▼KBcom.hand.hl$

components自定义message的发送工具类

►Blast

▼DirabbitMq

controllers发i]Json信息是producer

▼Bicomponents

QMessageSenderComponents

listener进行持久化处理、异常处理

▼Bicontrollers

QMessageSenderController

▼Ellistener

QContractLineLstener

QContractListener

0MessageConfrmCallBackListener

©MessageRetuinCallBackListener

3.1自定义发送类

利用AmqpTemplate的重载方法convertAndSendO方法，进行mq的发送，这个方法里有

三个参数,分别是exchange、router>message,即交换机、routing-key^json数据串。

在不同的功能里，有需要响应或者发送的就可以在无应的producer类里注入该类，然后

调用sendMessageO方法即可。

4

eM-g-omp。—.j--

packagecom.hand.his.rabbitMq.components;

Himportorg.springframevork.amqp.core.AmQpTemplate：

importorg.springframework.beans,factory,annotation.Autowired；

Zimportorg.springframework,stereotype.Component;

QComponent

publicclassMessageSenderComponents{

•Autovired

privateAmqpTempla:eamQpTemplate；

publicvoidsendMessage(Stringexchange.Stringrouter.Stringmessage){

amqpTemplate.convertAndSend(exchange,router,message):

AUKRDy、JOUIIJ1

3.2send类(producer)

以下面的发送为例。

首先定义两个常量EXCHANGE、ROUTER,即交换机，routing-key。(注意：要在mq的配置

文件里配置)见下图。

V!一交换机一>

<!~

交次机：一个交换机可以娟定多个队列，一个队列也可以娜定到3个交决机±

如果没有队列蜴定到交换机上.R1发送到谖交换机上的信息则会丢失

direct模式：消息与一个特定的跻由叁完全匹配，才会晴发

topic模式：技段则转发消息•灵活

—>

<rabblt:topic-exchangeid="ftXnaao・exchange・test"na«e=MRXngan.exchange.test**durable=Mtrue*'auto-delete='f：>

<rabbit:bindings>

<rabbit:bindingpattern^"sed*'queue,'fijjHifljj-queue_test,,/>

</rabbit:bindings>

7・ebbiI;lu/11y具

<rabbit:bindingpattern=Msod"qucuc="pinganqueuetestw/>

这个是为交换机绑定routing-key,一个交换机可以绑定多个routing-key,其中pattern属

性就是routing-key,queue属性就是queue。

5

package*coo.hand.his.rabbitMq.ccotroll^rs；

isport...

•Controller

Publicclass■飙出Sen如叙1以公班(

privatefinalStringEXCHVXE**pmiAn.exchange.test将exchange和router定义为常■以使下式引用

privatefinalStringROC7ER**sed>：

privateLoggerlogger•LogierFactory.r^r^wHgetClassO):

privateMtssageSenderCcapoTMntssend*rCooponents;IEA**11定义的发达关公在下用，。n-$3£。()

privateIPurContractStoreHetdServiceheadS*rvic«;一汴入业务类▼面“加长衰戍的增副改介的方法用门^override的方法

•RequestMappingCrabbitns(.send*)

requestMapplng可以方便前端Utff］或自己两试

^ResponseBody

publicStringf9ndO(

Li$t<PurContractStoreH^id>purContractStoreHeads-headServic^.0;

fo|(inti•0;i<pur€ontractStor^<cads.siz«();j44)(

//Stringsiss•JSONCbject.toJSOSStringCpurContractStoreHeads.get(i))：

Z/JSONObJectJson0b.eet■JSOKObject.perseObJectfJSONObject.toJS(^<purContrACtStoreHeads.get(i)).toStringO)：

//Strings=JsonObject.toStringO；

Stringsi-JSCAObjcct.ro/SttUpurContractStoreHeads.get(i>).toStrirgO：

s9nderCoopon«nts.serdMessag«(EXCHV»CE.ROUTER.3D;send方法体计对不同的业务需求“的

logger,info(sl):

j做是将从数据用出家的J・Q复合忖犊或Json对世法A；发送金之内

机.由到印wu。

return*HelloRabbitMq!*：

3.3监听类(consumer)

监听类要在mq配置文件里注册，监听类与queue有关，因为监听收到的json数据，会以

message的方式传到监听类里，然后对其进行处理。

<!一定义queue—>

49el­

sedurable是否持久化

51exclusive仅创建者可以使鹿的私有队列，断开后自动■除

52auto-delete当所有消弱建连接新开后,是苦自动■!除队列

53->

<rabbit:queueideMDinqaaQueuetestid"na«e=,,DloqaDqueuetest"durable:"true"auto-delete=a,falseMexclusive^"4alse,7>

注意；这里的queue和发送方定义的queue是一样的。

一般收到json数据以后，如果发送方对json数据进行了加密或者编码，则在收到json

数据以后，要先对json数据进行解密或者解码，然后在对■其进行json数据转换。

6

♦publicvoidonMessdge(MessageDtssage,Channelchannel)th-Exception(

JSONObjectj£QQffiu£Cl-null;

try(

oessage.getMessas^ocvrti«s().s«tCcntentTyp«(*appHcationjson*);将技校跑伯息更中&置为J»on：2

lotB*r.info(*c«tah7dBtssage:*♦Dossag«♦*fr»channtl:**channel：;

Objectobj-Dassas^Ccnv9rtor.frosM«5»>£e(xws$a£Q);■将转换或Java财象

jsonObject•JSOX.roJSttV$rrw(obj))；格J―川我转换成J/on咐效

//getdataandtransfertoJavaObject

PuiContracxSxoreHeadMad-JSOX.toJavaObJecz(JsonObject.PurConxractStoreHead.class)：将Jan"W.

//reoo^'eidfieldincaseduplicateerror

bead.setPurchas«<ontr8CtHead!d(null)：■&Kxd为null¥-AAH91Mlsequence

〃savetodatabaseandg6tanewid

PuKontractStoreHead{urContractStoreHead■h«adServ£ce.insertSelective：iRcqucst:null,head):a|*

lofger.debug(aReceiveaDessageparsetojson：{|*.puxContractStoreHead)

Icatch(Exceptionc)•

loftsor.Qrror(*Sca»error:*.•);

//errorSrvico.«tvoError<**.♦)：如果存ft鼻常信息9日定义一个“rvis

String—Body•*xx>*«>.setUessag^O:

//■aiIS*rvic*.s^ndMaiKa«ssateBody.*r^c^iv9r*«*proj*ct*.*cc*):

If(jsonObjecx!■null){

nessageSenderCoapawnTs.sendMessage(exchange：-wxchan«。"，router*ropic*.JaonObJect.toJSONStrlnf())：

〃必须手动ack才能将潸息从仄列中W抻

channel.basicAck(Dessage.o»tMessageProperties().getDeliveryTMO*&false):f动KIMIIl

I

4Mq的配置文件详解

4.1连接配置

这是mq配置文件的最基本和重要的配置，其中包括host：mq服务器的ip,userrame：

用户,password:密码，virtual-host：虚拟机。

消息确认监听器；是发送方用来监听json数据是否发送到交换机。见下图。

publicclassMessageConfirmCallBackListenerimplementsRabbitTemplate.ConfirmCallback{

privateLoggerlogger=LoggerFactory.getLogger(getClass());

11

(aOverride

publicvoidconfirm(CorrelationDatacorrelationData,booleanb,Strings){

if(b){

15//只确认生产者消息发送成功，消费者是否处理成功不做保证

logger.deoug("发送消息到exchange手工回应;

}else{

18//只确认生产者消息发送成功，消费者是否处理成功不做保证

logger,debug(“消息未发送到交换机”);

20)

21i>

22)

消息退回监听器：是发送方用来监听json数据是否从oxchango路由到queue。见下图。

publicclassMessageReturnCallBackListenerimplementsRabbitTemplate.Returncallback{

privateLoggerlogger=LoggerFactory.getlogger(getClass());

11

©Override

publicvoidreturnedMessage(Messagemessage,inti,Strings,Stringsi,Strings2){

logger,debug("exchange到queue不成功,回调return..;

15，}

16

7

<context:property-placeholderlocation-"classpath:config.propertit$*/>

。一连接配置一〉

<!一消息确认监听器一〉

<beanid-*confirtoCallBackListener*classec<mhand.his.rabbitMq.1.stener.MessageConfirtoCal!BackListenerV>

<：一消息退回监听器一》

<b^anid-rtturnCalIBackListtntr*classcom.hand.his.rabbitMQ.listtntr.MossdgeReturnCallBackListtntr*/^

<!一rabbit连接配置一〉

<rabbit:connection-factoryid-*connectionFactory*

publisher-confirms='true*

publisher-returns-*true*

host^^localhost*_RabbitMq股务器ip

username-Mfinance*

virtual-host="pa*J虚拟机叫用户最细粒度的权限

password_*finance.001*

port**5672*/>

4.2重试策略和消息转换器

消息转换器，要在mq配置文件里定义，这里用的是spring整合mq的消息转换器。

<1一■试策略一）

<b«anid**r«tryTesplat«*class**org.sprintfraserork.rttry.supp<

<properxy"beckOffPolicy*>

<be«nclass''»rg.spuntfranrrork.retry,backoff.Exp：

<propertynoi>®-'initialIntan-al-valur--5O0*/>

《propertynaue*iuitipner*value-iu.uV?

<propertynane-*a*xlnt«rval*value-*10000a/>

</bw>

</property>

</boan>

<!―消息婢展器.白格式一

<beanidJacksonSJsonMessAgcCon.crtcrclass

。一通il指定下面的adzin信息.当拿produc.r中的“chant♦粒会在rabbitnqfi务b上自动生或

<r«bbit:ediinCCOMCXlon*factarp*co«MCti«Nctory*A

4.3定义queue和交换机

不管是发送方还是接收方，都要配置queue,发送方要配置exchange、routing-key,接

收方要配置queue,因为发送方和接收方配置了相同的连接配置.（见3.1）,所以，mq服务器会

跟据配置文件，把发送方数据>exchange>routing-key>queue>监听类。

48<!―^xQueue—>

49<!-

5«durable是否持久化

51exclusive仅创建台可以使用的私府队列,断开后自就副球

52auto-delete当所有加重端连发新开后,是苦自动■!除队列

口一>

<rabbit:queueid«*DlttQan_Queue_testid"naa^^fiinoao-Queue-testMdurablerwauto-dele—“exclusive«wfalseM/>

<!一交涣机

57<!一

58交推机：一个交投机可以解定S个队列，一个队则也可以墀定到多个交接机上

59如果没有队列■定到交帙机上.照发送到该交换机上的信息JN会长笑

6e

61direct模式：簿息与一个特定陲由b究至匹髭，才会找发

62topic般式：接双则耨发消息，■艮送

63—>

<rabblt:topic-exchangeid11oAnaan.exchange,testHna»e=Holnflao.exchange.test"durable^:ueauto-delete=Mfalse">

<rabbit:bindings>

<rabbit:bindingpattern»,,sedMqueueMpAfiqafi_Queue_test,,/>

</rabbit:bindings>

</rabbit:topic-exchange〉

8

4.4监听器注册和队列配对

监听类，即写在listener包里的，要在配置文件里注册，然后在设置监听类与队列配对。

〈beanxdclass

〈beanidclass-

<rabbit:listener-container

connection-factory-

iwssate-converter'

acknowledte,nanual

<rabbit:listener

queues-rplnaaD-Q1

ref-*contractHea.

<rabbit:listener

queues-*pingan_qi

ref->contractLin-

</rabbit:Hsten«r-cootain«r>

5初始化开发环境（内网）

由于平安是内网环境，很多事情没有操作的权限，例如链接远程仓库下载Jar包依赖，对

于这种情况，需要由有经验的开发人员先确定好将来开发过程会使用到的依赖，避免在开发过

程中多次更新项目结构，减缓项目的实施进度。

5.1Git拉取远程仓库项目

BefdctorguildRunloohVCSlWindowHelp

springappli<otionC1nrMHktnry»Tomcat0.5.11▼A€

/VCSOperationsPopup...Alt*'

JCommit...CtrUK

/UpdateProject...CtrUT

-a-Pfo/ci...

RefreshFileStatus

手ShowLocalChangesasUMLCtrl*Alt-Shlft*D

Rt，

3CreatePatek.\+AddCtrHAIt*A

ApplyPatch...\八…、1•

ApplyPatchfiomClipboard...ShowCurrentRevision

己ShelveChanges...:Comparewithth6SameRepositoryVersion

ChectoutfromVersionControl►^omporewith...

ImportintoVersionControl►ComparewithBranch…

BrowseVCSRepository►Q)ShowHbtOfY

^ynr•与Revert...CtrkAtt♦Z

ResolveConflicts...

GotoFileCtrl+Sh

PBranches-CtrkShift*'

RecentFilesCtrl+ETag...

匕MergeCharges...

NavigationBarAltStashChanges...

UnSushChanges...

Dropfilesheretoc5ResetHEAD..

Remotes.^

/ML.

ZPush...Ctrl»$hift*K

Rebme...

9

SearchEverywhereDoubleShift

Droptilesheretoopen

iicloneH

5.2如何修改项目名

共需5步

①在idea里右击项目名选择Refactor—>Rename

②修改项目pom.xml的两处，见下图

<?xmlversion-^*1.0*encoding--ITF-8*?>

<projectxmlns^-http:naven.apache.org/POM/4.0.0

xalns:xsia*http://www.w3.org/2001/XMLSch<

xsi:schemaLocation-*http:/xaven.apache.(

<®odelVersion>4.0.0</modelVersion>

<groupld>com.hand.hls</groupld>

<artifactId>pazg<.artifactld>

<version>l.0.O-RELE.ASE</version>

<packaging>war</packaging>

<build>

<finalName>pazg</final\ame>

〈plugins)

plugin

<groupld>org.apache,maven.plugins</groupld>

<artifactld>maven-war-plugin</ar:ifactld>

<configuration>

③重新生成war

10

nProjectStructure

4-+

Name:pazg:war

6pazg：war

ProjectSettings

pazg:warexploded

Project

Outputdirectory:D:\Projects\pazg\t.

Modules