The following column Redis Tutorial will give you a detailed explanation of Redis and queues. I hope it will be helpful to friends in need!
Redis can not only be used as a cache server, but also as a message queue. Its list type inherently supports use as a message queue. As shown in the figure below:
Since the Redis list is implemented using a doubly linked list and saves the head and tail nodes, it is very fast to insert elements at the head and tail of the list. of.
So you can directly use Redis's List to implement the message queue, with just two simple instructions: lpush and rpop or rpush and lpop. A simple example is as follows:
Stores the message end (message producer):
package org.yamikaze.redis.messsage.queue;
import org.yamikaze.redis.test.MyJedisFactory;import redis.clients.jedis.Jedis;
import java.util.concurrent.TimeUnit;
/**
* 消息生产者
* @author yamikaze */public class Producer extends Thread {
public static final String MESSAGE_KEY = "message:queue"; private Jedis jedis; private String producerName; private volatile int count;
public Producer(String name) { this.producerName = name;
init();
}
private void init() {
jedis = MyJedisFactory.getLocalJedis();
}
public void putMessage(String message) {
Long size = jedis.lpush(MESSAGE_KEY, message);
System.out.println(producerName + ": 当前未被处理消息条数为:" + size);
count++;
}
public int getCount() { return count;
}
@Override public void run() { try { while (true) {
putMessage(StringUtils.generate32Str());
TimeUnit.SECONDS.sleep(1);
}
} catch (InterruptedException e) {
} catch (Exception e) {
e.printStackTrace();
}
}
public static void main(String[] args) throws InterruptedException{
Producer producer = new Producer("myProducer");
producer.start();
for(; ;) {
System.out.println("main : 已存储消息条数:" + producer.getCount());
TimeUnit.SECONDS.sleep(10);
}
}
}The message processing end (message consumer):
package org.yamikaze.redis.messsage.queue;
import org.yamikaze.redis.test.MyJedisFactory;import redis.clients.jedis.Jedis;
/**
* 消息消费者
* @author yamikaze */public class Customer extends Thread{
private String customerName; private volatile int count; private Jedis jedis;
public Customer(String name) { this.customerName = name;
init();
}
private void init() {
jedis = MyJedisFactory.getLocalJedis();
}
public void processMessage() {
String message = jedis.rpop(Producer.MESSAGE_KEY); if(message != null) {
count++;
handle(message);
}
}
public void handle(String message) {
System.out.println(customerName + " 正在处理消息,消息内容是: " + message + " 这是第" + count + "条");
}
@Override public void run() { while (true) {
processMessage();
}
}
public static void main(String[] args) {
Customer customer = new Customer("yamikaze");
customer.start();
}
}seems pretty good, but in the above example There is a problem with message consumers, that is, they need to constantly call the rpop method to check whether there are pending messages in the List. Each call will initiate a connection, which will cause unnecessary waste. Maybe you will use Thread.sleep() and other methods to let the consumer thread consume again after a period of time, but there are two problems in doing so:
1) If the producer speed is greater than the consumer consumption speed, the message queue The length will keep increasing and will occupy a lot of memory space over time.
2) If the sleep time is too long, some time-sensitive messages cannot be processed. If the sleep time is too short, it will also cause relatively large overhead on the connection.
So you can use the brpop instruction. This instruction will only return if there is an element. If not, it will block until timeout and return null. So the consumer can change the processMessage to this:
public void processMessage() { /**
* brpop支持多个列表(队列)
* brpop指令是支持队列优先级的，比如这个例子中MESSAGE_KEY的优先级大于testKey（顺序决定）。
* 如果两个列表中都有元素，会优先返回优先级高的列表中的元素，所以这儿优先返回MESSAGE_KEY
* 0表示不限制等待，会一直阻塞在这儿 */
List<String> messages = jedis.brpop(0, Producer.MESSAGE_KEY, "testKey"); if(messages.size() != 0) { //由于该指令可以监听多个Key,所以返回的是一个列表 //列表由2项组成，1) 列表名，2)数据
String keyName = messages.get(0); //如果返回的是MESSAGE_KEY的消息
if(Producer.MESSAGE_KEY.equals(keyName)) {
String message = messages.get(1);
handle(message);
}
}
System.out.println("=======================");
}Then You can run Customer and clear the console. You can see that the program has no output and is blocked here in brpop. Then open the Redis client and enter the command client list to see that there are currently two connections.
In addition to providing support for message queues, Redis also provides a set of commands to support the publish/subscribe mode. Using the pub/sub mode of Redis, you can implement a queue that produces once and consumes multiple times.
1) Publish
The PUBLISH instruction can be used to publish a message in the format PUBLISH channel message
The return value indicates the number of subscribers to the message.
2) Subscription
The SUBSCRIBE instruction is used to receive a message in the format SUBSCRIBE channel
You can see that after using the SUBSCRIBE instruction, you entered the subscription mode, but did not receive the message sent by publish. This is because Subscriptions will only receive the message before it is sent. For other commands in this mode, only replies can be seen. There are three types of replies:
1. If it is subscribe, the second value indicates the subscribed channel, and the third value indicates which channel is subscribed to? (understood as a serial number?)
2. If is message, the second value is the channel that generated the message, and the third value is message
3. If it is unsubscribe, the second value represents the unsubscribed channel, and the third value represents the current client's Number of subscriptions.
You can use the command UNSUBSCRIBE to unsubscribe. If no parameters are added, all channels subscribed to by the SUBSCRIBE command will be unsubscribed.
Redis also supports message subscription based on wildcards. Use the command PSUBSCRIBE (pattern subscribe), for example:
If you try to push the message again, you will get the following results:
You can see The publish command returns 2, and the subscriber received the message twice. This is because the PSUBSCRIBE command can subscribe to the channel repeatedly. Channels subscribed using the PSUBSCRIBE command must also be unsubscribed using the PUNSUBSCRIBE command. This command cannot unsubscribe from channels subscribed by SUBSCRIBE. Similarly, UNSUBSCRIBE cannot unsubscribe from channels subscribed by the PSUBSCRIBE command. At the same time, the PUNSUBSCRIBE instruction wildcard will not be expanded.
For example: PUNSUBSCRIBE * will not match channel.*, so to unsubscribe from channel.*, you need to write PUBSUBSCRIBE channel.* like this.
The code demonstration is as follows:
package org.yamikaze.redis.messsage.subscribe;
import org.yamikaze.redis.messsage.queue.StringUtils;import org.yamikaze.redis.test.MyJedisFactory;import redis.clients.jedis.Jedis;
/**
* 消息发布方
* @author yamikaze */public class Publisher {
public static final String CHANNEL_KEY = "channel:message"; private Jedis jedis;
public Publisher() {
jedis = MyJedisFactory.getLocalJedis();
}
public void publishMessage(String message) { if(StringUtils.isBlank(message)) { return;
}
jedis.publish(CHANNEL_KEY, message);
}
public static void main(String[] args) {
Publisher publisher = new Publisher();
publisher.publishMessage("Hello Redis!");
}
}Simply send a message.
Message Subscriber:
package org.yamikaze.redis.messsage.subscribe;
import org.yamikaze.redis.test.MyJedisFactory;import redis.clients.jedis.Jedis;import redis.clients.jedis.JedisPubSub;
import java.util.concurrent.TimeUnit;
/**
* 消息订阅方客户端
* @author yamikaze */public class SubscribeClient {
private Jedis jedis; private static final String EXIT_COMMAND = "exit";
public SubscribeClient() {
jedis = MyJedisFactory.getLocalJedis();
}
public void subscribe(String ...channel) { if(channel == null || channel.length <= 0) { return;
} //消息处理,接收到消息时如何处理
JedisPubSub jps = new JedisPubSub() { /**
* JedisPubSub类是一个没有抽象方法的抽象类,里面方法都是一些空实现
* 所以可以选择需要的方法覆盖,这儿使用的是SUBSCRIBE指令，所以覆盖了onMessage
* 如果使用PSUBSCRIBE指令，则覆盖onPMessage方法
* 当然也可以选择BinaryJedisPubSub,同样是抽象类，但方法参数为byte[] */
@Override public void onMessage(String channel, String message) { if(Publisher.CHANNEL_KEY.equals(channel)) {
System.out.println("接收到消息: channel : " + message); //接收到exit消息后退出
if(EXIT_COMMAND.equals(message)) {
System.exit(0);
}
}
}
/**
* 订阅时 */
@Override public void onSubscribe(String channel, int subscribedChannels) { if(Publisher.CHANNEL_KEY.equals(channel)) {
System.out.println("订阅了频道:" + channel);
}
}
}; //可以订阅多个频道 当前线程会阻塞在这儿 jedis.subscribe(jps, channel);
}
public static void main(String[] args) {
SubscribeClient client = new SubscribeClient();
client.subscribe(Publisher.CHANNEL_KEY); //并没有 unsubscribe方法 //相应的也没有punsubscribe方法 }
}First run the client, then run Publisher to send the message, the output result:
Redis Pub/sub also has its disadvantages, which is that if the consumer goes offline, the producer's messages will be lost.
In the process of business development, there will be some scenarios that require delay processing, such as:
a.订单下单之后超过30分钟用户未支付，需要取消订单
b.订单一些评论，如果48h用户未对商家评论，系统会自动产生一条默认评论
c.点我达订单下单后，超过一定时间订单未派出，需要超时取消订单等。。。
处理这类需求，比较直接简单的方式就是定时任务轮训扫表。这种处理方式在数据量不大的场景下是完全没问题，但是当数据量大的时候高频的轮训数据库就会比较的耗资源，导致数据库的慢查或者查询超时。所以在处理这类需求时候，采用了延时队列来完成。
延时队列就是一种带有延迟功能的消息队列。下面会介绍几种目前已有的延时队列：
优点：JDK自身实现，使用方便，量小适用
缺点：队列消息处于jvm内存，不支持分布式运行和消息持久化
优点：消息持久化，分布式
缺点：不支持任意时间精度，只支持特定level的延时消息
优点：消息持久化，分布式
缺点：延时相同的消息必须扔在同一个队列
Redis由于其自身的Zset数据结构，本质就是Set结构上加了个排序的功能，除了添加数据value之外，还提供另一属性score，这一属性在添加修改元素时候可以指定，每次指定后，Zset会自动重新按新的值调整顺序。可以理解为有两列字段的数据表，一列存value,一列存顺序编号。操作中key理解为zset的名字，那么对延时队列又有何用呢？
试想如果score代表的是想要执行时间的时间戳，在某个时间将它插入Zset集合中，它变会按照时间戳大小进行排序，也就是对执行时间前后进行排序，这样的话，起一个死循环线程不断地进行取第一个key值，如果当前时间戳大于等于该key值的socre就将它取出来进行消费删除，就可以达到延时执行的目的, 注意不需要遍历整个Zset集合，以免造成性能浪费。
Zset的排列效果如下图：
java代码实现如下：
package cn.chinotan.service.delayQueueRedis;import org.apache.commons.lang3.StringUtils;import redis.clients.jedis.Jedis;import redis.clients.jedis.JedisPool;import redis.clients.jedis.Tuple;import java.text.SimpleDateFormat;import java.util.Calendar;import java.util.Date;import java.util.Set;import java.util.concurrent.CountDownLatch;import java.util.concurrent.TimeUnit;/**
* @program: test
* @description: redis实现延时队列
* @author: xingcheng
* @create: 2018-08-19
**/public class AppTest { private static final String ADDR = "127.0.0.1"; private static final int PORT = 6379; private static JedisPool jedisPool = new JedisPool(ADDR, PORT); private static CountDownLatch cdl = new CountDownLatch(10); public static Jedis getJedis() { return jedisPool.getResource();
} /**
* 生产者,生成5个订单 */
public void productionDelayMessage() { for (int i = 0; i < 5; i++) {
Calendar instance = Calendar.getInstance(); // 3秒后执行
instance.add(Calendar.SECOND, 3 + i);
AppTest.getJedis().zadd("orderId", (instance.getTimeInMillis()) / 1000, StringUtils.join("000000000", i + 1));
System.out.println("生产订单: " + StringUtils.join("000000000", i + 1) + " 当前时间：" + new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date()));
System.out.println((3 + i) + "秒后执行");
}
} //消费者，取订单
public static void consumerDelayMessage() {
Jedis jedis = AppTest.getJedis(); while (true) {
Set<Tuple> order = jedis.zrangeWithScores("orderId", 0, 0); if (order == null || order.isEmpty()) {
System.out.println("当前没有等待的任务"); try {
TimeUnit.MICROSECONDS.sleep(500);
} catch (InterruptedException e) {
e.printStackTrace();
} continue;
}
Tuple tuple = (Tuple) order.toArray()[0]; double score = tuple.getScore();
Calendar instance = Calendar.getInstance(); long nowTime = instance.getTimeInMillis() / 1000; if (nowTime >= score) {
String element = tuple.getElement();
Long orderId = jedis.zrem("orderId", element); if (orderId > 0) {
System.out.println(new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date()) + ":redis消费了一个任务：消费的订单OrderId为" + element);
}
}
}
} static class DelayMessage implements Runnable{
@Override public void run() { try {
cdl.await();
consumerDelayMessage();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
public static void main(String[] args) {
AppTest appTest = new AppTest();
appTest.productionDelayMessage(); for (int i = 0; i < 10; i++) { new Thread(new DelayMessage()).start();
cdl.countDown();
}
}
}实现效果如下：
The above is the detailed content of Detailed explanation of Redis and queues. For more information, please follow other related articles on the PHP Chinese website!
Commonly used database software
What are the in-memory databases?
Which one has faster reading speed, mongodb or redis?
How to use redis as a cache server
How redis solves data consistency
How do mysql and redis ensure double-write consistency?
What data does redis cache generally store?
What are the 8 data types of redis
![[Web front-end] Node.js quick start](https://img.php.cn/upload/course/000/000/067/662b5d34ba7c0227.png)
