springboot整合redis实现消息队列

redis除了作为一般的nosql数据存储之外，还能实现简单的消息队列的功能。

项目结构


1.pom.xml如下

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
14.0.0

2

3   com.gwd

4   springboot-redis-provider_reciver

5   0.0.1-SNAPSHOT

6   jar

7

8   springboot-redis-provider_reciver

9   Demo project for springboot-provider_reciver

10

11   org.springframework.boot spring-boot-starter-parent 2.0.0.RELEASE 

12

13   UTF-8 UTF-8 1.8 

14

15   org.springframework.boot spring-boot-starter-data-redis org.springframework.boot spring-boot-starter-web org.springframework.boot spring-boot-starter-test test 

16

17   org.springframework.boot spring-boot-maven-plugin

18

1
2
1  2.properties文件 

2

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
1# REDIS (RedisProperties)

2# Redis数据库索引（默认为0）

3spring.redis.database=0

4# Redis服务器地址

5spring.redis.host=192.168.30.103

6# Redis服务器连接端口

7spring.redis.port=6379

8# Redis服务器连接密码（默认为空）

9spring.redis.password=

10# 连接池最大连接数（使用负值表示没有限制）

11spring.redis.jedis.pool.max-active=8

12# 连接池最大阻塞等待时间（使用负值表示没有限制）

13spring.redis.jedis.pool.max-wait=-1

14# 连接池中的最大空闲连接

15spring.redis.jedis.pool.max-idle=8

16# 连接池中的最小空闲连接

17spring.redis.jedis.pool.min-idle=0

18# 连接超时时间（毫秒）

19spring.redis.timeout=5000

20

21server.port=8088

22

3.Receiver.java

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
1package com.gwd.config;

2

3import java.util.concurrent.CountDownLatch;

4

5import org.springframework.beans.factory.annotation.Autowired;

6

7/** 

8* @FileName Receiver.java

9* @Description:TODO

10* @author JackHisen(gu.weidong)

11* @version V1.0

12* @createtime 2018年3月16日 下午4:44:08 

13* 修改历史：

14* 时间           作者          版本        描述

15*====================================================  

16*

17*/

18public class Receiver {

19  private CountDownLatch latch;

20

21    @Autowired

22    public Receiver(CountDownLatch latch) {

23        this.latch = latch;

24    }

25

26    public void receiveMessage(String message) {

27        System.out.println("Received <" + message + ">");

28        latch.countDown();

29    }

30}

31

4.启动类

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
1package com.gwd;

2

3import java.util.concurrent.CountDownLatch;

4

5import org.springframework.boot.SpringApplication;

6import org.springframework.boot.autoconfigure.SpringBootApplication;

7import org.springframework.context.annotation.Bean;

8import org.springframework.data.redis.connection.RedisConnectionFactory;

9import org.springframework.data.redis.core.StringRedisTemplate;

10import org.springframework.data.redis.listener.PatternTopic;

11import org.springframework.data.redis.listener.RedisMessageListenerContainer;

12import org.springframework.data.redis.listener.adapter.MessageListenerAdapter;

13

14import com.gwd.config.Receiver;

15

16@SpringBootApplication

17public class SpringbootRedisReciverApplication {

18

19  @Bean

20  Receiver receiver(CountDownLatch latch) {

21  return new Receiver(latch);

22  }

23   

24  //必要的redis消息队列连接工厂

25  @Bean

26  CountDownLatch latch() {

27  return new CountDownLatch(1);

28  }

29   

30  //redis模板

31  @Bean

32  StringRedisTemplate template(RedisConnectionFactory connectionFactory) {

33  return new StringRedisTemplate(connectionFactory);

34  }

35   

36  //注入消息监听器容器

37  @Bean

38  RedisMessageListenerContainer container(RedisConnectionFactory connectionFactory,MessageListenerAdapter listenerAdapter) {

39  RedisMessageListenerContainer container = new RedisMessageListenerContainer();

40  container.setConnectionFactory(connectionFactory);

41  container.addMessageListener(listenerAdapter, new PatternTopic("msg"));

42  return container;

43  }

44   

45  //注入消息监听器容器

46  @Bean

47  MessageListenerAdapter listenerAdapter(Receiver receiver) {

48  return new MessageListenerAdapter(receiver, "receiveMessage");

49  }

50  

51  

52  public static void main(String[] args) {

53      SpringApplication.run(SpringbootRedisReciverApplication.class, args);

54  }

55}

56

5.controller

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
1package com.gwd.controller;

2

3import java.util.concurrent.CountDownLatch;

4

5import org.springframework.beans.factory.annotation.Autowired;

6import org.springframework.data.redis.core.StringRedisTemplate;

7import org.springframework.web.bind.annotation.RequestMapping;

8import org.springframework.web.bind.annotation.ResponseBody;

9import org.springframework.web.bind.annotation.RestController;

10

11/** 

12* @FileName TestController.java

13* @Description:TODO

14* @author JackHisen(gu.weidong)

15* @version V1.0

16* @createtime 2018年3月16日 下午5:47:20 

17* 修改历史：

18* 时间           作者          版本        描述

19*====================================================  

20*

21*/

22@RestController

23public class TestController {

24  @Autowired

25  CountDownLatch latch;

26  @Autowired

27  StringRedisTemplate template;

28  

29  @RequestMapping("/testProvider")

30  @ResponseBody

31  public String testProvider() {

32      System.out.println("我要发送消息咯...");

33      template.convertAndSend("msg", "欢迎使用redis的消息队列!");

34      try {

35      //发送消息连接等待中

36          System.out.println("消息正在发送...");

37      latch.await();

38      } catch (InterruptedException e) {

39          System.out.println("消息发送失败...");

40      }

41      return null;

42  }

43}

44

1
2
1 6.测试结果： 

2


注意：配置文件中设置连接超时时间spring.redis.timeout不要为0，否则会报：

org.springframework.dao.QueryTimeoutException: Redis command timed out; nested exception is io.lettuce.core.RedisCommandTimeoutException: Command timed out