1、依赖
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
2、初始化配置
rabbitmq:
addresses: 10.0.0.236
port: 15672
username: root
password: 123456
virtual-host: /
listener:
simple:
concurrency: 10
max-concurrency: 10
prefetch: 1
auto-startup: true
default-requeue-rejected: true
template:
retry:
enabled: true
initial-interval: 1000
max-attempts: 3
max-interval: 10000
multiplier: 1
package dhht.seal.hn.gsgate.rabbitmq;
import org.springframework.amqp.core.Queue;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
/**
* @Author: sh
* @Description: RabbitMqConfig
* @Date: 10:33 2019/11/4
*/
@Configuration
public class RabbitMqConfig {
public static final String QUEUE = "gs_queue";
@Bean
public Queue cretaeQueue(){
return new Queue(QUEUE,true);
}
}
3、生产
package dhht.seal.hn.gsgate.rabbitmq;
import com.alibaba.fastjson.JSON;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
/**
* @Author: sh
* @Description: MqSender
* @Date: 10:34 2019/11/4
*/
@Slf4j
@Service
public class MqSenderService {
@Autowired
AmqpTemplate amqpTemplate;
public String sendMsgToQueue(Object message) {
String msg = beanToString(message);
log.info("sendMsgToQueue--messgae:" + msg);
//amqpTemplate.convertAndSend(MQConfig.QUEUE, msg);
return amqpTemplate.convertSendAndReceive(RabbitMqConfig.QUEUE, message).toString();
}
public void sendMsg(Object message) {
String msg = beanToString(message);
log.info("send message:" + msg);
amqpTemplate.convertAndSend(RabbitMqConfig.QUEUE, msg);
log.info("sendMsg（）---消息发送成功！");
}
public static <T> String beanToString(T value) {
if (value == null) {
return null;
}
Class<?> clazz = value.getClass();
if (clazz == int.class || clazz == Integer.class) {
return "" + value;
} else if (clazz == String.class) {
return (String) value;
} else if (clazz == long.class || clazz == Long.class) {
return "" + value;
} else {
return JSON.toJSONString(value);
}
}
}
4、消费
package dhht.seal.hn.gsgate.rabbitmq;
import dhht.seal.hn.gsgate.service.CropQueryService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.messaging.handler.annotation.SendTo;
import org.springframework.stereotype.Service;
import javax.annotation.Resource;
/**
* @Author: sh
* @Description: MqReceiver
* @Date: 10:40 2019/11/4
*/
@Service
public class MqReceiverService {
private static Logger log = LoggerFactory.getLogger(MqReceiverService.class);
@Resource
private CropQueryService cropQueryService;
@RabbitListener(queues = RabbitMqConfig.QUEUE)
@SendTo(RabbitMqConfig.QUEUE)
public String receiveQueueMsg(String message) {
log.info("接收到队列消息：" + message);
// 业务处理代码，工商拉取入库
String resJson = cropQueryService.crropQuery(message);
return resJson;
}
}