码迷,mamicode.com
首页 > 编程语言 > 详细

SpringBoot整合RabbitMQ,实现消息发送和消费

时间:2018-12-05 17:25:23      阅读:204      评论:0      收藏:0      [点我收藏+]

标签:junit   字符   core   tor   otv   slf4j   tar   ESS   .com   

下载安装Erlang和RabbitMQ

Erlang和RabbitMQ:https://www.cnblogs.com/theRhyme/p/10069611.html

 

项目创建和依赖

推荐SpringCloud项目在线创建:https://start.spring.io/

不用上面这个也行,下面有代码和依赖;

gradle的依赖,和maven差不多:

buildscript {
    ext {
        springBootVersion = ‘2.1.1.RELEASE‘
    }
    repositories {
        mavenCentral()
    }
    dependencies {
        classpath("org.springframework.boot:spring-boot-gradle-plugin:${springBootVersion}")
    }
}

apply plugin: ‘java‘
apply plugin: ‘eclipse‘
apply plugin: ‘org.springframework.boot‘
apply plugin: ‘io.spring.dependency-management‘

group = ‘xy.study‘
version = ‘0.0.1-SNAPSHOT‘
sourceCompatibility = 1.8

repositories {
    mavenCentral()
}


dependencies {
    implementation(‘org.springframework.boot:spring-boot-starter-amqp‘)
    implementation(‘org.projectlombok:lombok:1.16.+‘)
    runtimeOnly(‘org.springframework.boot:spring-boot-devtools‘)
    testImplementation(‘org.springframework.boot:spring-boot-starter-test‘)
}

 

代码

配置文件application.properties

spring.application.name=spring-boot-rabbitmq

spring.rabbitmq.host=localhost
spring.rabbitmq.port=5672
spring.rabbitmq.username=guest
spring.rabbitmq.password=guest



server.port = 5678

 

RabbitMQ配置文件类(注释的代码可以直接删掉):

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;


/**
 * topic 是RabbitMQ中最灵活的一种方式,可以根据routing_key自由的绑定不同的队列
 * 首先对topic规则配置
 */
//@Configuration
public class TopicRabbitConfig {

    final public static String QUEUE_NAME = "queue.name";
    final public static String TEST_TOPIC_ROUTINGKEY = "test.topic.routingKey";
    final public static String TEST_EXCHANGE_HAHA = "test.exchange.haha";



    /**
     * 设置交换器的名称
     * @return
     *//*
    @Bean
    TopicExchange exchange() {
        return new TopicExchange(TopicRabbitConfig.TEST_EXCHANGE_HAHA);
    }

    *//**
     * 队列名称
     * @return
     *//*
    @Bean
    public Queue queueMessage() {
        return new Queue(TopicRabbitConfig.QUEUE_NAME);
    }

    *//**
     * 将指定routing key的名称绑定交换器的队列
     * @param queueMessage
     * @param exchange
     * @return
     *//*
    @Bean
    Binding bindingExchangeMessage(Queue queueMessage, TopicExchange exchange) {
        return BindingBuilder.bind(queueMessage).to(exchange).with(TopicRabbitConfig.TEST_TOPIC_ROUTINGKEY);
    }*/

    /**
     * 匹配以topic开头的路由键routing key
     * 交换机绑定多个队列
     */

    /*@Bean
    Binding bindingExchangeMessages(Queue queueMessages, TopicExchange exchange) {
        return BindingBuilder.bind(queueMessages).to(exchange).with("topic.#");
    }*/
}

 

 生产者,这里根据Exchange和Routing Key,直接发送一个字符串:

import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import xy.study.rabbitmq.conf.TopicRabbitConfig;

@Component
@Slf4j
public class HelloSender {

    @Autowired
    private RabbitTemplate rabbitTemplate;


    /**
     * 通过exchange和routingKey的方式
     * rabbitTemplate.convertAndSend(String exchange, String routingKey, Object object)
     * @param i
     */
    public void send(int i) {
        String context = "hello " + i;
        log.info("Sender : {}", context);
        this.rabbitTemplate.convertAndSend(TopicRabbitConfig.TEST_EXCHANGE_HAHA,TopicRabbitConfig.TEST_TOPIC_ROUTINGKEY, context);
    }
}

 

消费者,绑定对应的Exchange,Queue和Routing Key,直接打印获取的信息:

import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.ExchangeTypes;
import org.springframework.amqp.rabbit.annotation.Exchange;
import org.springframework.amqp.rabbit.annotation.Queue;
import org.springframework.amqp.rabbit.annotation.QueueBinding;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.stereotype.Component;
import xy.study.rabbitmq.conf.TopicRabbitConfig;

@Component
@Slf4j
public class HelloReceiver {
    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(value = TopicRabbitConfig.QUEUE_NAME, durable = "true"),
            exchange = @Exchange(value = TopicRabbitConfig.TEST_EXCHANGE_HAHA, type = ExchangeTypes.TOPIC),
            key = TopicRabbitConfig.TEST_TOPIC_ROUTINGKEY)
    )
    public void onOrgDeleted(@Payload String msg) {
        log.info("HelloReceiver msg : {}",msg);
    }
}

 

测试类,调用生产者发送信息的函数send,消费者会监听消费:

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.SpringRunner;
import xy.study.rabbitmq.producer.HelloSender;

@RunWith(SpringRunner.class)
@SpringBootTest
public class RabbitmqApplicationTests {

    @Autowired
    private HelloSender sender;

    @Test
    public void testSend() {
        sender.send(666);
    }

}

 

如图,控制台日志,能生成消息,并且能被对应的消费者消费。

技术分享图片

 

 

 

 

SpringBoot整合RabbitMQ,实现消息发送和消费

标签:junit   字符   core   tor   otv   slf4j   tar   ESS   .com   

原文地址:https://www.cnblogs.com/theRhyme/p/10071781.html

(0)
(0)
   
举报
评论 一句话评论(0
登录后才能评论!
© 2014 mamicode.com 版权所有  联系我们:gaon5@hotmail.com
迷上了代码!