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:
  application:
    name: 3d-gis
  datasource:
    driver-class-name: org.postgresql.Driver
    url: jdbc:postgresql://172.16.10.201:5432/3d_gis
    username: postgres
    password: postgres
  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();
    

RabbitSender类

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.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 message 	具体的消息内容
     * @throws Exception
     */
    public void send(Object message)  
        CorrelationData data = new CorrelationData(UUID.randomUUID().toString());
        rabbitTemplate.convertAndSend(RabbitMQConfig.GIS_GRAPHICS_EXCHANGE, RabbitMQConfig.GIS_DATA_SEND_QUEUE, message, data);
    

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

RabbitReceiver类

import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import lombok.extern.slf4j.Slf4j;
import org.springblade.gis.modules.db.service.DbDataService;
import org.springframework.amqp.rabbit.annotation.*;
import org.springframework.amqp.core.Message;
import org.springframework.stereotype.Component;
import com.rabbitmq.client.Channel;

import javax.annotation.Resource;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;

@Component
@Slf4j
public class RabbitReceiver 

    @Resource
    private DbDataService dbDataService;
    @Resource
    private RabbitSender sender;



    /**
     * 	组合使用监听
     * @param message
     * @param channel
     */
    @RabbitListener(queues =RabbitMQConfig.GIS_DATA_RECEIVE_REQUEUE)
    public void onMessage(String msg,Message message, Channel channel) 
        //	1. 收到消息以后进行业务端消费处理
        String body=new String(message.getBody());
        log.info("消费消息:" + body );
        JSONObject json;
        try 
            json=JSONObject.parseObject(body);
         catch (Exception e) 
            ackOrReject(message,channel,true);
            return;
        
        String table= json.getString("table");
        JSONArray data= json.getJSONArray("data");
        List<Map<String,Object>> maps=new ArrayList<>();
        data.forEach(e->
           Map<String,Object> map=(Map<String, Object>) e;
            maps.add(map);
         );
         List<Long> ids=dbDataService.saveBatch(table,maps);
         sender.send(JSONArray.toJSONString(ids));
        //  2. 处理成功之后 获取deliveryTag 并进行手工的ACK操作, 因为我们配置文件里配置的是 手工签收
         ackOrReject(message,channel,true);
    

    private void ackOrReject(Message message, Channel channel, boolean result)  
        try 
            if (result) 
                channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
             else 
                channel.basicReject(message.getMessageProperties().getDeliveryTag(), false);
            
        catch (IOException e)
            new IOException();
        
    

测试类

import io.swagger.annotations.Api;
import io.swagger.annotations.ApiOperation;
import lombok.AllArgsConstructor;
import org.springblade.core.secure.annotation.NoToken;
import org.springblade.core.tool.api.R;
import org.springblade.three.dimension.gis.rabbitmq.RabbitSender;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import javax.annotation.Resource;

@NoToken
@RestController
@AllArgsConstructor
@RequestMapping("rabbit")
@Api(value = "消息队列", tags = "消息队列")
public class RabbitController 
    @Resource
    private RabbitSender sender;


    @ApiOperation(value = "添加 @author Tarzan Liu")
    @GetMapping("test")
    public R test()
        sender.send("测试666");
        return R.status(true);
    

测试结果