Redis消息队列与订阅者发布者
2018-10-16 本文已影响0人
代码potty
Redis列表List是采用的双端链表的结构，所有头尾存取元素特别快
image.png
手动操作一下在java中实现redis的消息队列，通过生产者和消费者的模式进行实现
生产者代码：
public class Producer extends Thread{
public static final String messageKey = "message";
private JedisPool jedisPool;
private String producerName;
private Jedis jedis;
private volatile int count = 0;
public Producer(String producerName){
this.producerName = producerName;
jedisPool = new JedisPool("127.0.0.1",6379);//redis的ip和端口
jedis = jedisPool.getResource();
}
public void putMessage(String message){
jedis.lpush(messageKey,message);
count++;
}
public void run(){
while (true){
try {
putMessage(RandomStringUtils.randomAlphabetic(5));//获取长度为5的任意字符串
TimeUnit.SECONDS.sleep(1);//线程进入为时一秒的休眠状态
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
public int getCount(){
return count;
}
public static void main(String[] args) throws InterruptedException {
Producer producer = new Producer("myTest");
producer.start();
for (;;){
System.out.println("生产者:"+producer.producerName+"生产了"+producer.getCount()+"对象");
TimeUnit.SECONDS.sleep(10);
}
}
}
消费者代码
public class Customer extends Thread{
private String customerName;
private volatile int count = 0;
private Jedis jedis;
private JedisPool jedisPool;
public Customer(String customerName){
this.customerName = customerName;
jedisPool = new JedisPool("127.0.0.1",6379);
jedis = jedisPool.getResource();
}
//注意点
public void getMessage(){
List<String> result = jedis.blpop(0,Producer.messageKey);
if (result.size()!=0){
String rkey = result.get(0);
if (rkey.equals(Producer.messageKey)){
String msg = result.get(1);
count++;
showMessage(msg);
}
}
}
public void showMessage(String msg){
System.out.println("消费者:"+customerName+"消息内容是"+msg+"这是第"+count+"个产品");
}
public void run(){
while (true){
getMessage();
}
}
public static void main(String[] args) {
Customer customer = new Customer("消费一号");
customer.start();
}
}
其中比较重要的方法就是getMessage(),可以看到我使用的是blpop()的方式取出消息的，因为blpop是一个阻塞的方法，当消息队列中没有元素，那么就会一直进入到阻塞的状态，如果使用lpop的方式，那么就会一直进行尝试，会浪费很多的资源，得不偿失，所以在此使用拥有阻塞功能的blpop()方法可以提供资源的利用率。
redis还提供了订阅者和发布者的功能，类似于QQ聊天，可以私聊，群聊的方式。
发布者代码
public class Publisher {
public static final String channel = "aChannel";
private JedisPool jedisPool;
private Jedis jedis;
private String pname;
public Publisher(String pname){
this.pname = pname;
jedisPool = new JedisPool("127.0.0.1",6379);
jedis = jedisPool.getResource();
}
public void publish(String msg){
if (org.apache.commons.lang.StringUtils.isBlank(msg))
return;
jedis.publish(channel,msg);
}
public static void main(String[] args) {
Publisher publisher = new Publisher("发布者");
publisher.publish("cececcee");
}
}
这个代码块主要注意在方法publish中，发布消息的时候，消息不能为空，如果传入空值字符串则直接返回。
订阅者
public class Subscriber {
private final String exitInfo = "exit";
private JedisPool jedisPool;
private Jedis jedis;
private String subscriber;
public Subscriber(String subscriber){
this.subscriber = subscriber;
jedisPool = new JedisPool("127.0.0.1",6379);
jedis = jedisPool.getResource();
}
public void subscribe(String ...channel){
if (channel==null || channel.length<=0)
return;
JedisPubSub jedisPubSub = new JedisPubSub() {
@Override
public void onMessage(String channel, String message) {
if (Publisher.channel.equals(channel)){
System.out.println("接收到频道:"+channel+"的消息:"+message);
if (message.equals(exitInfo)){
System.exit(0);
}
}
}
@Override
public void onSubscribe(String channel, int subscribedChannels) {
if (Publisher.channel.equals(channel)){
System.out.println("订阅了频道:"+channel);
}
}
};
jedis.subscribe(jedisPubSub,channel);
}
public static void main(String[] args) {
Subscriber subscriber = new Subscriber("订阅者");
subscriber.subscribe(Publisher.channel);
}
}
订阅者中，主要注意JedisPubSub这个类，JedisPubSub类是一个没有抽象方法的抽象类,里面方法都是一些空实现，所以我们可以选择性的去重写某些方法来实现我们的目的，在代码块中可以看出我重写了onMessage()方法和onSubscribe()方法，前一个方法是在接收数据时候的处理，后一个方法是在订阅频道时候的处理，最后再调用jedis的subscribe()方法来实现订阅频道。
image.png
image.png
经过测试，成功发送和接收信息。
总结
使用Redis的List数据结构可以简单迅速地做一个消息队列，同时Redis提供的BRPOP和BLPOP等指令解决了频繁调用Jedis的rpop和lpop方法造成的资源浪费问题。除此之外，Redis提供对发布/订阅模式的指令，可以实现消息传递、进程间通信。
补充说明随机字符串实现类
//产生5位长度的随机字符串，中文环境下是乱码
RandomStringUtils.random(5);
//使用指定的字符生成5位长度的随机字符串
RandomStringUtils.random(5, new char[]{'a','b','c','d','e','f', '1', '2', '3'});
//生成指定长度的字母和数字的随机组合字符串
RandomStringUtils.randomAlphanumeric(5);
//生成随机数字字符串
RandomStringUtils.randomNumeric(5);
//生成随机[a-z]字符串，包含大小写
RandomStringUtils.randomAlphabetic(5);
//生成从ASCII 32到126组成的随机字符串
RandomStringUtils.randomAscii(4)
想要使用这个类的方法需要导入相应的包
<dependency>
<groupId>org.apache.directory.studio</groupId>
<artifactId>org.apache.commons.lang</artifactId>
<version>2.6</version>
</dependency>
参考链接：https://blog.csdn.net/qq_34212276/article/details/78455004
