基于dubbo的分布式项目框架搭建 开发工具idea (springboot+dubbo+zookeeper+redis+rabbitmq+基于Swagger2的restful api) --(四)

1.rabbitmq的集成

首先在配置文件里增加

[code]#rabbitMQ
spring.rabbitmq.host=127.0.0.1
spring.rabbitmq.port=5672
spring.rabbitmq.username=root
spring.rabbitmq.password=root
spring.rabbitmq.publisher-confirms=true
spring.rabbitmq.publisher-returns=true
spring.rabbitmq.template.mandatory=true

1.1这里的username和password是需要接下来我们自己在rabbitmq里添加的用户和密码，也可以使用默认的用户和密码guest

1.2 安装rabbitmq之前需要先安装OTP 点击进入以下网址，自己选择合适版本下载

http://www.erlang.org/downloads

安装完后继续安装rabbitmq

http://www.rabbitmq.com/download.html

 rabbitmq 和 otp 全都默认下一步安装即可

详细的步骤请看

https://www.geek-share.com/detail/2685650960.html

安装配置完成后，进入管控台界面，可以在queues里看到声明的队列的信息

配置队列等信息

import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.core.TopicExchange;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

/**
* @author hhp
* @Title: RabbitConfiguration
* @Description: mq 队列，配置
* @return:
* @throws
*/
@Configuration
public class RabbitConfiguration {

// 声明队列
@Bean
public Queue queue1() {
return new Queue("hello.queue1", true);
}

@Bean
public Queue queue2() {
return new Queue("hello.queue2", true);
}

// 声明交互器
@Bean
TopicExchange topicExchange() {
return new TopicExchange("topicExchange");
}

// 绑定
@Bean
public Binding binding1() {
return BindingBuilder.bind(queue1()).to(topicExchange()).with("key.1");
}

@Bean
public Binding binding2() {
return BindingBuilder.bind(queue2()).to(topicExchange()).with("key.#");
}

}

配置消息的发送者

import org.apache.log4j.Logger;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.core.RabbitTemplate.ReturnCallback;
import org.springframework.amqp.rabbit.support.CorrelationData;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.util.UUID;

/**
* @author hhp
* @Title: Sender
* @Description: 向mq发送消息
* @param:
* @return:
* @throws
*/
@Component
public class Sender implements RabbitTemplate.ConfirmCallback, ReturnCallback {

private static Logger logger = Logger.getLogger(App.class);

@Autowired
private RabbitTemplate rabbitTemplate;

@PostConstruct
public void init() {
rabbitTemplate.setConfirmCallback(this);
rabbitTemplate.setReturnCallback(this);
}

@Override
public void confirm(CorrelationData correlationData, boolean ack, String cause) {
if (ack) {
logger.info("消息发送成功:" + correlationData);
} else {
logger.info("消息发送失败:" + cause);
}

}

@Override
public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
logger.error(message.getMessageProperties().getCorrelationIdString() + " 发送失败");

}

// 发送消息，
public void send(String msg) {

CorrelationData correlationId = new CorrelationData(UUID.randomUUID().toString());

logger.info("开始发送消息 : " + msg.toLowerCase());
String response = rabbitTemplate.convertSendAndReceive("topicExchange", "key.1", msg, correlationId).toString();
logger.info("结束发送消息 : " + msg.toLowerCase());
logger.info("消费者响应 : " + response + " 消息处理完成");
}
}

配置消息的接收者

/**
* @author hhp
* @Title: Receiver
* @Description: 接收消息
* @param:
* @return:
* @throws
*/
@Component
public class Receiver {

private static Logger logger = Logger.getLogger(App.class);

@RabbitListener(queues = "hello.queue1")
public String processMessage1(String msg) {
logger.info(Thread.currentThread().getName() + " 接收到来自hello.queue1队列的消息：" + msg);
return msg.toUpperCase();
}

@RabbitListener(queues = "hello.queue2")
public void processMessage2(String msg) {
logger.info(Thread.currentThread().getName() + " 接收到来自hello.queue2队列的消息：" + msg);
}
}

运行test类

import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;

import com.ysk.component.Sender;

@RunWith(value = SpringJUnit4ClassRunner.class)
@SpringBootTest(classes = App.class)
public class RabbitTest {

@Autowired
private Sender sender;

@Test
public void sendTest() throws Exception {
for (int i = 0; i < 3; i++) {
sender.send("消息发送测试");
}
}
}