Springboot项目使用Rabbitmq同步调用

Posted 2021-12-14 洛阳泰山

首先在springboot的pom文件里引入依赖

<dependency>
	<groupId>org.springframework.boot</groupId>
	<artifactId>spring-boot-starter-amqp</artifactId>
</dependency>

然后application.yml：

ps：里面的虚拟host配置项不是必须的，我自己在rabbitmq服务上创建了自己的虚拟host，所以我配置了；你们不创建，就不用加这个配置项。

spring:
  rabbitmq:
    publisher-confirms: true
    publisher-returns: true
    host: 172.16.10.201
    port: 5672
    username: leaniot
    password: leaniot
    virtual-host: /3d_gis
    listener:
      simple:
        acknowledge-mode: manual
      direct:
        acknowledge-mode: manual

RabbitMQ配置类

 
import org.springframework.amqp.core.*;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
 
 
 
/**
 * RabbitMQ 配置
 *
 * @author tarzan
 * @version 1.0
 * @date 2021年05月18日 09:15:40
 * @since JDK1.8
 */
 
@Configuration
public class RabbitMQConfig
 
    /** gis图形交换机 */
    public static final String  GIS_GRAPHICS_EXCHANGE = "3d_gis_exchange";
    /** gis图形数据接收序列 */
    public static final String GIS_DATA_RECEIVE_REQUEUE = "gis_data_queue";
    /** gis图形数据返回（发送）序列 */
    public static final String GIS_DATA_SEND_QUEUE = "gis_result_queue";
 
    @Bean
    public Queue gisDataReceiveQueue () 
        return new Queue(GIS_DATA_RECEIVE_REQUEUE);
    
 
    @Bean
    public Queue gisDataSendQueue () 
        return new Queue(GIS_DATA_SEND_QUEUE);
    
 
 
    @Bean
    public DirectExchange directExchange() 
        return new DirectExchange(GIS_GRAPHICS_EXCHANGE);
    
 
    @Bean
    public Binding receiveBinding () 
        return BindingBuilder.bind(gisDataReceiveQueue()).to(directExchange()).withQueueName();
    
 
    @Bean
    public Binding sendBinding () 
        return BindingBuilder.bind(gisDataSendQueue()).to(directExchange()).withQueueName();
    
 
 
 

消息生产者

import java.nio.charset.StandardCharsets;
import java.util.UUID;

import lombok.extern.slf4j.Slf4j;
import org.springblade.gis.rabbitmq.confirm.RabbitReturnCallback;
import org.springblade.gis.rabbitmq.confirm.RabbitConfirmCallback;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;

@Component
@Slf4j
public class RabbitSender implements InitializingBean 

    @Resource
    private RabbitTemplate rabbitTemplate;

    /**
     * 	异步 对外发送消息的方法
     * @param msgId &msg	具体的消息内容
     * @throws Exception
     */
    public void syncSend(String msgId,String msg)  
        MessageProperties messageProperties=new MessageProperties();
        messageProperties.setMessageId(msgId);
        Message message=new Message(msg.getBytes(StandardCharsets.UTF_8),messageProperties);
        CorrelationData data = new CorrelationData(UUID.randomUUID().toString());
        rabbitTemplate.convertAndSend(RabbitMQConfig.GIS_GRAPHICS_EXCHANGE, RabbitMQConfig.GIS_DATA_SEND_QUEUE, message,data);
    

    /**
     *  同步	对外发送消息的方法
     * @param msgId &msg	具体的消息内容
     * @throws Exception
     */
    public Object asyncSend(String msgId,String msg) 
            MessageProperties messageProperties=new MessageProperties();
            messageProperties.setMessageId(msgId);
            Message message=new Message(msg.getBytes(StandardCharsets.UTF_8),messageProperties);
            CorrelationData data = new CorrelationData(UUID.randomUUID().toString());
            log.info("【发送的消息-社会信用代码】：" + msg);
            Object mm = rabbitTemplate.convertSendAndReceive(RabbitMQConfig.GIS_GRAPHICS_EXCHANGE,RabbitMQConfig.GIS_DATA_SEND_QUEUE, message,data);
            log.info("【同步消息返回结果-msgResult】："+mm.toString());
            return mm;

    

    @Override
    public void afterPropertiesSet () 
        rabbitTemplate.setConfirmCallback(new RabbitConfirmCallback());
        rabbitTemplate.setReturnCallback(new RabbitReturnCallback());
        rabbitTemplate.setMandatory(true);
    

消息消费者

import com.alibaba.fastjson.JSONObject;
import lombok.AllArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springblade.core.tool.utils.Func;
import org.springframework.amqp.rabbit.annotation.*;
import org.springframework.amqp.core.Message;
import org.springframework.messaging.handler.annotation.SendTo;
import org.springframework.stereotype.Component;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;

@Component
@Slf4j
@AllArgsConstructor
public class RabbitReceiver 


    /**
     * 	组合使用监听
     * @param message
     * @param channel
     */
    @RabbitListener(queues =RabbitMQConfig.GIS_DATA_RECEIVE_QUEUE)
    @SendTo(RabbitMQConfig.GIS_DATA_RECEIVE_QUEUE)
    public String receiveQueueMsg(Message message) 
        //	1. 收到消息以后进行业务端消费处理
        log.info("消息id:" + message.getMessageProperties().getMessageId() );
       String messageId= message.getMessageProperties().getMessageId();
        Long userId=0l;
        JSONObject json;
        try 
         userId= Func.toLong(messageId);
        String body=new String(message.getBody());
        log.info("消费消息:" + body );
         json=JSONObject.parseObject(body);
         catch (Exception e) 
            return "error";
        
         return "6666";
    

测试controller

import com.alibaba.fastjson.JSONObject;
import io.swagger.annotations.Api;
import io.swagger.annotations.ApiOperation;
import lombok.AllArgsConstructor;
import org.springblade.core.secure.utils.AuthUtil;
import org.springblade.core.tool.api.R;
import org.springblade.core.tool.utils.IoUtil;
import org.springblade.gis.modules.feature.data.dto.DataDTO;
import org.springblade.gis.rabbitmq.RabbitSender;
import org.springframework.web.bind.annotation.*;

import javax.annotation.Resource;
import javax.validation.Valid;
import java.io.File;
import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.util.Arrays;


@RestController
@AllArgsConstructor
@RequestMapping("rabbit")
@Api(value = "消息队列", tags = "消息队列")
public class RabbitController 

    @Resource
    private RabbitSender sender;


    @ApiOperation(value = "添加文件数据 @author Tarzan Liu")
    @PostMapping("importFile")
    public R importFile(@RequestBody @Valid DataDTO dto) 
        dto.setFuncKey("InputDB");
        sender.asyncSend(AuthUtil.getUserId().toString(),JSONObject.toJSONString(dto));
        return R.status(true);