Java編程rabbitMQ實(shí)現(xiàn)消息的收發(fā)
更新時(shí)間：2017年09月13日 17:21:55 作者：sdyy321
RabbitMQ是一個(gè)在AMQP基礎(chǔ)上完成的，可復(fù)用的企業(yè)消息系統(tǒng)，本文通過(guò)實(shí)例來(lái)給大家分享通過(guò)操作rabbitMQ實(shí)現(xiàn)消息的收發(fā)，感興趣的朋友可以參考下。
java實(shí)現(xiàn)rAMQP，即Advanced Message Queuing Protocol，高級(jí)消息隊(duì)列協(xié)議，是應(yīng)用層協(xié)議的一個(gè)開(kāi)放標(biāo)準(zhǔn)，為面向消息的中間件設(shè)計(jì)。消息中間件主要用于組件之間的解耦，消息的發(fā)送者無(wú)需知道消息使用者的存在，反之亦然。
AMQP的主要特征是面向消息、隊(duì)列、路由（包括點(diǎn)對(duì)點(diǎn)和發(fā)布/訂閱）、可靠性、安全。
RabbitMQ是一個(gè)開(kāi)源的AMQP實(shí)現(xiàn)，服務(wù)器端用Erlang語(yǔ)言編寫，支持多種客戶端，如：Python、Ruby、.NET、Java、JMS、C、PHP、ActionScript、XMPP、STOMP等，支持AJAX。用于在分布式系統(tǒng)中存儲(chǔ)轉(zhuǎn)發(fā)消息，在易用性、擴(kuò)展性、高可用性等方面表現(xiàn)不俗。
本文不介紹amqp和rabbitmq相關(guān)知識(shí)，請(qǐng)自行網(wǎng)上查閱
本文是基于spring-rabbit中間件來(lái)實(shí)現(xiàn)消息的發(fā)送接受功能
see http://www.rabbitmq.com/tutorials/tutorial-one-java.html
see http://www.springsource.org/spring-amqp
Java編程通過(guò)操作rabbitMQ消息的收發(fā)實(shí)現(xiàn)代碼如下：
<!-- for rabbitmq --> <dependency> <groupId>com.rabbitmq</groupId> <artifactId>amqp-client</artifactId> <version>2.8.2</version> </dependency> <dependency> <groupId>org.springframework.amqp</groupId> <artifactId>spring-amqp</artifactId> <version>1.1.1.RELEASE</version> </dependency> <dependency> <groupId>org.springframework.amqp</groupId> <artifactId>spring-rabbit</artifactId> <version>1.1.1.RELEASE</version> </dependency> <dependency> <groupId>com.caucho</groupId> <artifactId>hessian</artifactId> <version>4.0.7</version> </dependency> </dependencies>
首先我們需要一個(gè)用來(lái)在app和rabbitmq之間傳遞消息的持有對(duì)象
public class EventMessage implements Serializable{
private String queueName;
private String exchangeName;
private byte[] eventData;
public EventMessage(String queueName, String exchangeName, byte[] eventData) {
this.queueName = queueName;
this.exchangeName = exchangeName;
this.eventData = eventData;
}
public EventMessage() {
}
public String getQueueName() {
return queueName;
}
public String getExchangeName() {
return exchangeName;
}
public byte[] getEventData() {
return eventData;
}
@Override
public String toString() {
return "EopEventMessage [queueName=" + queueName + ", exchangeName="
+ exchangeName + ", eventData=" + Arrays.toString(eventData)
+ "]";
}
}
為了可以發(fā)送和接受這個(gè)消息持有對(duì)象，我們還需要需要一個(gè)用來(lái)序列化和反序列化的工廠
public interface CodecFactory {
byte[] serialize(Object obj) throws IOException;
Object deSerialize(byte[] in) throws IOException;
}
下面是編碼解碼的實(shí)現(xiàn)類，用了hessian來(lái)實(shí)現(xiàn)，大家可以自行選擇序列化方式
public class HessionCodecFactory implements CodecFactory {
private final Logger logger = Logger.getLogger(HessionCodecFactory.class);
@Override
public byte[] serialize(Object obj) throws IOException {
ByteArrayOutputStream baos = null;
HessianOutput output = null;
try {
baos = new ByteArrayOutputStream(1024);
output = new HessianOutput(baos);
output.startCall();
output.writeObject(obj);
output.completeCall();
} catch (final IOException ex) {
throw ex;
} finally {
if (output != null) {
try {
baos.close();
} catch (final IOException ex) {
this.logger.error("Failed to close stream.", ex);
}
}
}
return baos != null ? baos.toByteArray() : null;
}
@Override
public Object deSerialize(byte[] in) throws IOException {
Object obj = null;
ByteArrayInputStream bais = null;
HessianInput input = null;
try {
bais = new ByteArrayInputStream(in);
input = new HessianInput(bais);
input.startReply();
obj = input.readObject();
input.completeReply();
} catch (final IOException ex) {
throw ex;
} catch (final Throwable e) {
this.logger.error("Failed to decode object.", e);
} finally {
if (input != null) {
try {
bais.close();
} catch (final IOException ex) {
this.logger.error("Failed to close stream.", ex);
}
}
}
return obj;
}
}
接下來(lái)就先實(shí)現(xiàn)發(fā)送功能，新增一個(gè)接口專門用來(lái)實(shí)現(xiàn)發(fā)送功能
public interface EventTemplate {
void send(String queueName,String exchangeName,Object eventContent) throws SendRefuseException;
void send(String queueName,String exchangeName,Object eventContent,CodecFactory codecFactory) throws SendRefuseException;
}
SendRefuseException是自定義的發(fā)送失敗異常類
下面是它的實(shí)現(xiàn)類，主要的任務(wù)就是將數(shù)據(jù)轉(zhuǎn)換為EventMessage
public class DefaultEventTemplate implements EventTemplate {
private static final Logger logger = Logger.getLogger(DefaultEventTemplate.class);
private AmqpTemplate eventAmqpTemplate;
private CodecFactory defaultCodecFactory;
// private DefaultEventController eec;
// public DefaultEventTemplate(AmqpTemplate eopAmqpTemplate,
// CodecFactory defaultCodecFactory, DefaultEventController eec) {
// this.eventAmqpTemplate = eopAmqpTemplate;
// this.defaultCodecFactory = defaultCodecFactory;
// this.eec = eec;
// }
public DefaultEventTemplate(AmqpTemplate eopAmqpTemplate,CodecFactory defaultCodecFactory) {
this.eventAmqpTemplate = eopAmqpTemplate;
this.defaultCodecFactory = defaultCodecFactory;
}
@Override
public void send(String queueName, String exchangeName, Object eventContent)
throws SendRefuseException {
this.send(queueName, exchangeName, eventContent, defaultCodecFactory);
}
@Override
public void send(String queueName, String exchangeName, Object eventContent,
CodecFactory codecFactory) throws SendRefuseException {
if (StringUtils.isEmpty(queueName) || StringUtils.isEmpty(exchangeName)) {
throw new SendRefuseException("queueName exchangeName can not be empty.");
}
// if (!eec.beBinded(exchangeName, queueName))
// eec.declareBinding(exchangeName, queueName);
byte[] eventContentBytes = null;
if (codecFactory == null) {
if (eventContent == null) {
logger.warn("Find eventContent is null,are you sure...");
} else {
throw new SendRefuseException(
"codecFactory must not be null ,unless eventContent is null");
}
} else {
try {
eventContentBytes = codecFactory.serialize(eventContent);
} catch (IOException e) {
throw new SendRefuseException(e);
}
}
// 構(gòu)造成Message
EventMessage msg = new EventMessage(queueName, exchangeName,
eventContentBytes);
try {
eventAmqpTemplate.convertAndSend(exchangeName, queueName, msg);
} catch (AmqpException e) {
logger.error("send event fail. Event Message : [" + eventContent + "]", e);
throw new SendRefuseException("send event fail", e);
}
}
}
注釋的地方稍后會(huì)用到，主要是防止數(shù)據(jù)數(shù)據(jù)發(fā)送的地方?jīng)]有事先聲明
然后我們?cè)賹?shí)現(xiàn)接受消息
首先我們需要一個(gè)消費(fèi)接口，所有的消費(fèi)程序都實(shí)現(xiàn)這個(gè)類
public interface EventProcesser {
public void process(Object e);
}
為了能夠?qū)⒉煌愋偷南⒔挥蓪?duì)應(yīng)的程序來(lái)處理，我們還需要一個(gè)消息處理適配器
/**
* MessageListenerAdapter的Pojo
* <p>消息處理適配器，主要功能：</p>
* <p>1、將不同的消息類型綁定到對(duì)應(yīng)的處理器并本地緩存，如將queue01+exchange01的消息統(tǒng)一交由A處理器來(lái)出來(lái)</p>
* <p>2、執(zhí)行消息的消費(fèi)分發(fā)，調(diào)用相應(yīng)的處理器來(lái)消費(fèi)屬于它的消息</p>
*
*/
public class MessageAdapterHandler {
private static final Logger logger = Logger.getLogger(MessageAdapterHandler.class);
private ConcurrentMap<String, EventProcessorWrap> epwMap;
public MessageAdapterHandler() {
this.epwMap = new ConcurrentHashMap<String, EventProcessorWrap>();
}
public void handleMessage(EventMessage eem) {
logger.debug("Receive an EventMessage: [" + eem + "]");
// 先要判斷接收到的message是否是空的，在某些異常情況下，會(huì)產(chǎn)生空值
if (eem == null) {
logger.warn("Receive an null EventMessage, it may product some errors, and processing message is canceled.");
return;
}
if (StringUtils.isEmpty(eem.getQueueName()) || StringUtils.isEmpty(eem.getExchangeName())) {
logger.warn("The EventMessage's queueName and exchangeName is empty, this is not allowed, and processing message is canceled.");
return;
}
// 解碼，并交給對(duì)應(yīng)的EventHandle執(zhí)行
EventProcessorWrap eepw = epwMap.get(eem.getQueueName()+"|"+eem.getExchangeName());
if (eepw == null) {
logger.warn("Receive an EopEventMessage, but no processor can do it.");
return;
}
try {
eepw.process(eem.getEventData());
} catch (IOException e) {
logger.error("Event content can not be Deserialized, check the provided CodecFactory.",e);
return;
}
}
protected void add(String queueName, String exchangeName, EventProcesser processor,CodecFactory codecFactory) {
if (StringUtils.isEmpty(queueName) || StringUtils.isEmpty(exchangeName) || processor == null || codecFactory == null) {
throw new RuntimeException("queueName and exchangeName can not be empty,and processor or codecFactory can not be null. ");
}
EventProcessorWrap epw = new EventProcessorWrap(codecFactory,processor);
EventProcessorWrap oldProcessorWrap = epwMap.putIfAbsent(queueName + "|" + exchangeName, epw);
if (oldProcessorWrap != null) {
logger.warn("The processor of this queue and exchange exists, and the new one can't be add");
}
}
protected Set<String> getAllBinding() {
Set<String> keySet = epwMap.keySet();
return keySet;
}
protected static class EventProcessorWrap {
private CodecFactory codecFactory;
private EventProcesser eep;
protected EventProcessorWrap(CodecFactory codecFactory,
EventProcesser eep) {
this.codecFactory = codecFactory;
this.eep = eep;
}
public void process(byte[] eventData) throws IOException{
Object obj = codecFactory.deSerialize(eventData);
eep.process(obj);
}
}
}
這是正常情況下的消息處理方式，如果rabbitmq消息接受發(fā)生異常，也要監(jiān)控到，新增一個(gè)消費(fèi)類專門用來(lái)處理錯(cuò)誤異常的消息
public class MessageErrorHandler implements ErrorHandler{
private static final Logger logger = Logger.getLogger(MessageErrorHandler.class);
@Override
public void handleError(Throwable t) {
logger.error("RabbitMQ happen a error:" + t.getMessage(), t);
}
}
接下來(lái)我們可能需要一個(gè)專門配置和rabbitmq通信的一些信息，比如地址，端口等信息
public class EventControlConfig {
private final static int DEFAULT_PORT = 5672;
private final static String DEFAULT_USERNAME = "guest";
private final static String DEFAULT_PASSWORD = "guest";
private final static int DEFAULT_PROCESS_THREAD_NUM = Runtime.getRuntime().availableProcessors() * 2;
private static final int PREFETCH_SIZE = 1;
private String serverHost ;
private int port = DEFAULT_PORT;
private String username = DEFAULT_USERNAME;
private String password = DEFAULT_PASSWORD;
private String virtualHost;
/**
* 和rabbitmq建立連接的超時(shí)時(shí)間
*/
private int connectionTimeout = 0;
/**
* 事件消息處理線程數(shù)，默認(rèn)是 CPU核數(shù) * 2
*/
private int eventMsgProcessNum;
/**
* 每次消費(fèi)消息的預(yù)取值
*/
private int prefetchSize;
public EventControlConfig(String serverHost) {
this(serverHost,DEFAULT_PORT,DEFAULT_USERNAME,DEFAULT_PASSWORD,null,0,DEFAULT_PROCESS_THREAD_NUM,DEFAULT_PROCESS_THREAD_NUM,new HessionCodecFactory());
}
public EventControlConfig(String serverHost, int port, String username,
String password, String virtualHost, int connectionTimeout,
int eventMsgProcessNum,int prefetchSize,CodecFactory defaultCodecFactory) {
this.serverHost = serverHost;
this.port = port>0?port:DEFAULT_PORT;
this.username = username;
this.password = password;
this.virtualHost = virtualHost;
this.connectionTimeout = connectionTimeout>0?connectionTimeout:0;
this.eventMsgProcessNum = eventMsgProcessNum>0?eventMsgProcessNum:DEFAULT_PROCESS_THREAD_NUM;
this.prefetchSize = prefetchSize>0?prefetchSize:PREFETCH_SIZE;
}
public String getServerHost() {
return serverHost;
}
public int getPort() {
return port;
}
public String getUsername() {
return username;
}
public String getPassword() {
return password;
}
public String getVirtualHost() {
return virtualHost;
}
public int getConnectionTimeout() {
return connectionTimeout;
}
public int getEventMsgProcessNum() {
return eventMsgProcessNum;
}
public int getPrefetchSize() {
return prefetchSize;
}
}
具體的發(fā)送、接受程序已經(jīng)好了，接下來(lái)也是最重要的就是管理控制和rabbitmq的通信
public interface EventController {
/**
* 控制器啟動(dòng)方法
*/
void start();
/**
* 獲取發(fā)送模版
*/
EventTemplate getEopEventTemplate();
/**
* 綁定消費(fèi)程序到對(duì)應(yīng)的exchange和queue
*/
EventController add(String queueName, String exchangeName, EventProcesser eventProcesser);
/*in map, the key is queue name, but value is exchange name*/
EventController add(Map<String,String> bindings, EventProcesser eventProcesser);
}
它的實(shí)現(xiàn)類如下：
/**
* 和rabbitmq通信的控制器，主要負(fù)責(zé)：
* <p>1、和rabbitmq建立連接</p>
* <p>2、聲明exChange和queue以及它們的綁定關(guān)系</p>
* <p>3、啟動(dòng)消息監(jiān)聽(tīng)容器，并將不同消息的處理者綁定到對(duì)應(yīng)的exchange和queue上</p>
* <p>4、持有消息發(fā)送模版以及所有exchange、queue和綁定關(guān)系的本地緩存</p>
* @author yangyong
*
*/
public class DefaultEventController implements EventController {
private CachingConnectionFactory rabbitConnectionFactory;
private EventControlConfig config;
private RabbitAdmin rabbitAdmin;
private CodecFactory defaultCodecFactory = new HessionCodecFactory();
private SimpleMessageListenerContainer msgListenerContainer; // rabbitMQ msg listener container
private MessageAdapterHandler msgAdapterHandler = new MessageAdapterHandler();
private MessageConverter serializerMessageConverter = new SerializerMessageConverter(); // 直接指定
//queue cache, key is exchangeName
private Map<String, DirectExchange> exchanges = new HashMap<String,DirectExchange>();
//queue cache, key is queueName
private Map<String, Queue> queues = new HashMap<String, Queue>();
//bind relation of queue to exchange cache, value is exchangeName | queueName
private Set<String> binded = new HashSet<String>();
private EventTemplate eventTemplate; // 給App使用的Event發(fā)送客戶端
private AtomicBoolean isStarted = new AtomicBoolean(false);
private static DefaultEventController defaultEventController;
public synchronized static DefaultEventController getInstance(EventControlConfig config){
if(defaultEventController==null){
defaultEventController = new DefaultEventController(config);
}
return defaultEventController;
}
private DefaultEventController(EventControlConfig config){
if (config == null) {
throw new IllegalArgumentException("Config can not be null.");
}
this.config = config;
initRabbitConnectionFactory();
// 初始化AmqpAdmin
rabbitAdmin = new RabbitAdmin(rabbitConnectionFactory);
// 初始化RabbitTemplate
RabbitTemplate rabbitTemplate = new RabbitTemplate(rabbitConnectionFactory);
rabbitTemplate.setMessageConverter(serializerMessageConverter);
eventTemplate = new DefaultEventTemplate(rabbitTemplate,defaultCodecFactory, this);
}
/**
* 初始化rabbitmq連接
*/
private void initRabbitConnectionFactory() {
rabbitConnectionFactory = new CachingConnectionFactory();
rabbitConnectionFactory.setHost(config.getServerHost());
rabbitConnectionFactory.setChannelCacheSize(config.getEventMsgProcessNum());
rabbitConnectionFactory.setPort(config.getPort());
rabbitConnectionFactory.setUsername(config.getUsername());
rabbitConnectionFactory.setPassword(config.getPassword());
if (!StringUtils.isEmpty(config.getVirtualHost())) {
rabbitConnectionFactory.setVirtualHost(config.getVirtualHost());
}
}
/**
* 注銷程序
*/
public synchronized void destroy() throws Exception {
if (!isStarted.get()) {
return;
}
msgListenerContainer.stop();
eventTemplate = null;
rabbitAdmin = null;
rabbitConnectionFactory.destroy();
}
@Override
public void start() {
if (isStarted.get()) {
return;
}
Set<String> mapping = msgAdapterHandler.getAllBinding();
for (String relation : mapping) {
String[] relaArr = relation.split("\\|");
declareBinding(relaArr[1], relaArr[0]);
}
initMsgListenerAdapter();
isStarted.set(true);
}
/**
* 初始化消息監(jiān)聽(tīng)器容器
*/
private void initMsgListenerAdapter(){
MessageListener listener = new MessageListenerAdapter(msgAdapterHandler,serializerMessageConverter);
msgListenerContainer = new SimpleMessageListenerContainer();
msgListenerContainer.setConnectionFactory(rabbitConnectionFactory);
msgListenerContainer.setAcknowledgeMode(AcknowledgeMode.AUTO);
msgListenerContainer.setMessageListener(listener);
msgListenerContainer.setErrorHandler(new MessageErrorHandler());
msgListenerContainer.setPrefetchCount(config.getPrefetchSize()); // 設(shè)置每個(gè)消費(fèi)者消息的預(yù)取值
msgListenerContainer.setConcurrentConsumers(config.getEventMsgProcessNum());
msgListenerContainer.setTxSize(config.getPrefetchSize());//設(shè)置有事務(wù)時(shí)處理的消息數(shù)
msgListenerContainer.setQueues(queues.values().toArray(new Queue[queues.size()]));
msgListenerContainer.start();
}
@Override
public EventTemplate getEopEventTemplate() {
return eventTemplate;
}
@Override
public EventController add(String queueName, String exchangeName,EventProcesser eventProcesser) {
return add(queueName, exchangeName, eventProcesser, defaultCodecFactory);
}
public EventController add(String queueName, String exchangeName,EventProcesser eventProcesser,CodecFactory codecFactory) {
msgAdapterHandler.add(queueName, exchangeName, eventProcesser, defaultCodecFactory);
if(isStarted.get()){
initMsgListenerAdapter();
}
return this;
}
@Override
public EventController add(Map<String, String> bindings,
EventProcesser eventProcesser) {
return add(bindings, eventProcesser,defaultCodecFactory);
}
public EventController add(Map<String, String> bindings,
EventProcesser eventProcesser, CodecFactory codecFactory) {
for(Map.Entry<String, String> item: bindings.entrySet())
msgAdapterHandler.add(item.getKey(),item.getValue(), eventProcesser,codecFactory);
return this;
}
/**
* exchange和queue是否已經(jīng)綁定
*/
protected boolean beBinded(String exchangeName, String queueName) {
return binded.contains(exchangeName+"|"+queueName);
}
/**
* 聲明exchange和queue已經(jīng)它們的綁定關(guān)系
*/
protected synchronized void declareBinding(String exchangeName, String queueName) {
String bindRelation = exchangeName+"|"+queueName;
if (binded.contains(bindRelation)) return;
boolean needBinding = false;
DirectExchange directExchange = exchanges.get(exchangeName);
if(directExchange == null) {
directExchange = new DirectExchange(exchangeName, true, false, null);
exchanges.put(exchangeName, directExchange);
rabbitAdmin.declareExchange(directExchange);//聲明exchange
needBinding = true;
}
Queue queue = queues.get(queueName);
if(queue == null) {
queue = new Queue(queueName, true, false, false);
queues.put(queueName, queue);
rabbitAdmin.declareQueue(queue); //聲明queue
needBinding = true;
}
if(needBinding) {
Binding binding = BindingBuilder.bind(queue).to(directExchange).with(queueName);//將queue綁定到exchange
rabbitAdmin.declareBinding(binding);//聲明綁定關(guān)系
binded.add(bindRelation);
}
}
}
搞定，現(xiàn)在可以將DefaultEventTemplate里的注釋去掉了，接下來(lái)最后完成單元測(cè)試，為了測(cè)試傳遞對(duì)象，建立一個(gè)PO
@SuppressWarnings("serial")
public class People implements Serializable{
private int id;
private String name;
private boolean male;
private People spouse;
private List<People> friends;
public int getId() {
return id;
}
public void setId(int id) {
this.id = id;
}
public String getName() {
return name;
}
public void setName(String name) {
this.name = name;
}
public boolean isMale() {
return male;
}
public void setMale(boolean male) {
this.male = male;
}
public People getSpouse() {
return spouse;
}
public void setSpouse(People spouse) {
this.spouse = spouse;
}
public List<People> getFriends() {
return friends;
}
public void setFriends(List<People> friends) {
this.friends = friends;
}
@Override
public String toString() {
// TODO Auto-generated method stub
return "People[id="+id+",name="+name+",male="+male+"]";
}
}
建立單元測(cè)試
public class RabbitMqTest{
private String defaultHost = "127.0.0.1";
private String defaultExchange = "EXCHANGE_DIRECT_TEST";
private String defaultQueue = "QUEUE_TEST";
private DefaultEventController controller;
private EventTemplate eventTemplate;
@Before
public void init() throws IOException{
EventControlConfig config = new EventControlConfig(defaultHost);
controller = DefaultEventController.getInstance(config);
eventTemplate = controller.getEopEventTemplate();
controller.add(defaultQueue, defaultExchange, new ApiProcessEventProcessor());
controller.start();
}
@Test
public void sendString() throws SendRefuseException{
eventTemplate.send(defaultQueue, defaultExchange, "hello world");
}
@Test
public void sendObject() throws SendRefuseException{
eventTemplate.send(defaultQueue, defaultExchange, mockObj());
}
@Test
public void sendTemp() throws SendRefuseException, InterruptedException{
String tempExchange = "EXCHANGE_DIRECT_TEST_TEMP";//以前未聲明的exchange
String tempQueue = "QUEUE_TEST_TEMP";//以前未聲明的queue
eventTemplate.send(tempQueue, tempExchange, mockObj());
//發(fā)送成功后此時(shí)不會(huì)接受到消息，還需要綁定對(duì)應(yīng)的消費(fèi)程序
controller.add(tempQueue, tempExchange, new ApiProcessEventProcessor());
}
@After
public void end() throws InterruptedException{
Thread.sleep(2000);
}
private People mockObj(){
People jack = new People();
jack.setId(1);
jack.setName("JACK");
jack.setMale(true);
List<People> friends = new ArrayList<>();
friends.add(jack);
People hanMeiMei = new People();
hanMeiMei.setId(1);
hanMeiMei.setName("韓梅梅");
hanMeiMei.setMale(false);
hanMeiMei.setFriends(friends);
People liLei = new People();
liLei.setId(2);
liLei.setName("李雷");
liLei.setMale(true);
liLei.setFriends(friends);
liLei.setSpouse(hanMeiMei);
hanMeiMei.setSpouse(liLei);
return hanMeiMei;
}
class ApiProcessEventProcessor implements EventProcesser{
@Override
public void process(Object e) {//消費(fèi)程序這里只是打印信息
Assert.assertNotNull(e);
System.out.println(e);
if(e instanceof People){
People people = (People)e;
System.out.println(people.getSpouse());
System.out.println(people.getFriends());
}
}
}
}
源碼地址請(qǐng)點(diǎn)擊這里
總結(jié)
以上就是本文關(guān)于java實(shí)現(xiàn)rabbitmq消息的發(fā)送接受的全部?jī)?nèi)容，希望對(duì)大家有所幫助。
感謝大家對(duì)本站的支持。
相關(guān)文章
mybatis如何使用Java8的日期LocalDate和LocalDateTime詳解
這篇文章主要給大家介紹了關(guān)于mybatis如何使用Java8的日期LocalDate和LocalDateTime的相關(guān)資料，文中通過(guò)示例代碼介紹的非常詳細(xì)，對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值，需要的朋友們下面隨著小編來(lái)一起學(xué)習(xí)學(xué)習(xí)吧。2017-09-09
Struts 2中實(shí)現(xiàn)Ajax的三種方式
這篇文章主要介紹了Struts 2中實(shí)現(xiàn)Ajax的三種方式，本文通過(guò)實(shí)例代碼給大家介紹的非常詳細(xì)，具有一定的參考借鑒價(jià)值,需要的朋友可以參考下2019-05-05
Java設(shè)計(jì)模式 模板模式及應(yīng)用場(chǎng)景解析
這篇文章主要介紹了Java設(shè)計(jì)模式 模板模式及應(yīng)用場(chǎng)景解析,文中通過(guò)示例代碼介紹的非常詳細(xì)，對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值,需要的朋友可以參考下2019-07-07
spring+maven實(shí)現(xiàn)發(fā)送郵件功能
這篇文章主要為大家詳細(xì)介紹了spring+maven實(shí)現(xiàn)發(fā)送郵件功能，利用spring提供的郵件工具來(lái)發(fā)送郵件，具有一定的參考價(jià)值，感興趣的小伙伴們可以參考一下2018-07-07
Spring Boot 項(xiàng)目創(chuàng)建的詳細(xì)步驟(圖文)
這篇文章主要介紹了Spring Boot 項(xiàng)目創(chuàng)建的詳細(xì)步驟(圖文)，這里我們有兩種創(chuàng)建Spring Boot項(xiàng)目的方法，具有一定的參考價(jià)值，感興趣的小伙伴們可以參考一下2019-05-05
SpringBoot實(shí)現(xiàn)二維碼掃碼登錄的原理及項(xiàng)目實(shí)踐
本文主要介紹了SpringBoot實(shí)現(xiàn)二維碼掃碼登錄的原理及項(xiàng)目實(shí)踐，文中通過(guò)示例代碼介紹的非常詳細(xì)，對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值，需要的朋友們下面隨著小編來(lái)一起學(xué)習(xí)學(xué)習(xí)吧2023-04-04
SpringBoot應(yīng)用部署到Tomcat中無(wú)法啟動(dòng)的解決方法
這篇文章主要介紹了SpringBoot應(yīng)用部署到Tomcat中無(wú)法啟動(dòng)的解決方法,需要的朋友可以參考下2017-09-09
很多小伙伴不知道如何記錄日志,今天特地整理了本篇文章,文中有非常詳細(xì)的介紹及代碼示例,對(duì)小伙伴們很有幫助,需要的朋友可以參考下2021-06-06
Mybatis基于MapperScan注解的動(dòng)態(tài)代理加載機(jī)制詳解
這篇文章主要介紹了Mybatis基于MapperScan注解的動(dòng)態(tài)代理加載機(jī)制，文中通過(guò)示例代碼介紹的非常詳細(xì)，對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值，需要的朋友們下面隨著小編來(lái)一起學(xué)習(xí)吧2023-01-01
關(guān)于Idea中的.properties文件顯示問(wèn)題
這篇文章主要介紹了關(guān)于Idea中的.properties文件顯示問(wèn)題,具有很好的參考價(jià)值,希望對(duì)大家有所幫助,如有錯(cuò)誤或未考慮完全的地方,望不吝賜教2024-07-07