首页 / REDIS / 详解Redis和队列

详解Redis和队列

内容导读

互联网集市收集整理的这篇技术教程文章主要介绍了详解Redis和队列，小编现在分享给大家，供广大互联网技能从业者学习和参考。文章包含12139字，纯文字阅读大概需要18分钟。

内容图文

下面由Redis教程栏目给大家详解Redis和队列，希望对需要的朋友有所帮助！

详解Redis和队列 - 文章图片

概要

Redis不仅可作为缓存服务器，还可用作消息队列。它的列表类型天生支持用作消息队列。如下图所示：

详解Redis和队列 - 文章图片

由于Redis的列表是使用双向链表实现的，保存了头尾节点，所以在列表头尾两边插取元素都是非常快的。

普通队列实现

所以可以直接使用Redis的List实现消息队列，只需简单的两个指令lpush和rpop或者rpush和lpop。简单示例如下：

存放消息端(消息生产者):

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);
        }
    }
}

消息处理端(消息消费者):

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();
    }
}

貌似还不错，但上述例子中消息消费者有一个问题存在，即需要不停的调用rpop方法查看List中是否有待处理消息。每调用一次都会发起一次连接，这会造成不必要的浪费。也许你会使用Thread.sleep()等方法让消费者线程隔一段时间再消费，但这样做有两个问题：

1）、如果生产者速度大于消费者消费速度，消息队列长度会一直增大，时间久了会占用大量内存空间。

2）、如果睡眠时间过长，这样不能处理一些时效性的消息，睡眠时间过短，也会在连接上造成比较大的开销。

所以可以使用brpop指令，这个指令只有在有元素时才返回，没有则会阻塞直到超时返回null，于是消费端可以将processMessage可以改为这样：

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("=======================");
}

然后可以运行Customer，清空控制台，可以看到程序没有任何输出，阻塞在了brpop这儿。然后在打开Redis的客户端，输入指令client list，可以查看当前有两个连接。

一次生产多次消费的队列

Redis除了对消息队列提供支持外，还提供了一组命令用于支持发布/订阅模式。利用Redis的pub/sub模式可以实现一次生产多次消费的队列。

1)发布
PUBLISH指令可用于发布一条消息，格式 PUBLISH channel message

返回值表示订阅了该消息的数量。
2)订阅
SUBSCRIBE指令用于接收一条消息，格式 SUBSCRIBE channel

可以看到使用SUBSCRIBE指令后进入了订阅模式，但没有接收到publish发送的消息，这是因为只有在消息发出去前订阅才会接收到。在这个模式下其他指令，只能看到回复。回复分为三种类型：
1、如果为subscribe，第二个值表示订阅的频道，第三个值表示是第几个订阅的频道?(理解成序号?)
2、如果为message(消息)，第二个值为产生该消息的频道，第三个值为消息
3、如果为unsubscribe，第二个值表示取消订阅的频道，第三个值表示当前客户端的订阅数量。

可以使用指令UNSUBSCRIBE退订，如果不加参数，则会退订所有由SUBSCRIBE指令订阅的频道。

Redis还支持基于通配符的消息订阅，使用指令PSUBSCRIBE (pattern subscribe)，例如：

再试试推送消息会得到以下结果：

可以看到publish指令返回的是2，而订阅端这边接收了两次消息。这是因为PSUBSCRIBE指令可以重复订阅频道。而使用PSUBSCRIBE指令订阅的频道也要使用指令PUNSUBSCRIBE指令退订，该指令无法退订SUBSCRIBE订阅的频道，同理UNSUBSCRIBE也不能退订PSUBSCRIBE指令订阅的频道。同时PUNSUBSCRIBE指令通配符不会展开。
例如：PUNSUBSCRIBE * 不会匹配到 channel.*, 所以要取消订阅channel.*就要这样写PUBSUBSCRIBE channel.*。

代码示范如下：

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!");
    }
}

简单的发送一个消息。

消息订阅方：

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方法    }
}

先运行client，再运行Publisher进行消息发送，输出结果：

详解Redis和队列 - 文章图片

Redis的pub/sub也有其缺点，那就是如果消费者下线，生产者的消息会丢失。

延时队列

背景

在业务发展过程中，会出现一些需要延时处理的场景，比如：

a.订单下单之后超过30分钟用户未支付，需要取消订单
b.订单一些评论，如果48h用户未对商家评论，系统会自动产生一条默认评论
c.点我达订单下单后，超过一定时间订单未派出，需要超时取消订单等。。。
处理这类需求，比较直接简单的方式就是定时任务轮训扫表。这种处理方式在数据量不大的场景下是完全没问题，但是当数据量大的时候高频的轮训数据库就会比较的耗资源，导致数据库的慢查或者查询超时。所以在处理这类需求时候，采用了延时队列来完成。

几种延时队列

延时队列就是一种带有延迟功能的消息队列。下面会介绍几种目前已有的延时队列：

1.Java中java.util.concurrent.DelayQueue

优点：JDK自身实现，使用方便，量小适用
缺点：队列消息处于jvm内存，不支持分布式运行和消息持久化

2.Rocketmq延时队列

优点：消息持久化，分布式
缺点：不支持任意时间精度，只支持特定level的延时消息

3.Rabbitmq延时队列（TTL+DLX实现）

优点：消息持久化，分布式
缺点：延时相同的消息必须扔在同一个队列

Redis实现的延时消息队列适合的项目特点：

Spring框架管理对象
有消息需求，但不想维护mq中间件
有使用redis
对消息持久化并没有很苛刻的要求

Redis实现的延时消息队列思路

Redis由于其自身的Zset数据结构，本质就是Set结构上加了个排序的功能，除了添加数据value之外，还提供另一属性score，这一属性在添加修改元素时候可以指定，每次指定后，Zset会自动重新按新的值调整顺序。可以理解为有两列字段的数据表，一列存value,一列存顺序编号。操作中key理解为zset的名字，那么对延时队列又有何用呢？

试想如果score代表的是想要执行时间的时间戳，在某个时间将它插入Zset集合中，它变会按照时间戳大小进行排序，也就是对执行时间前后进行排序，这样的话，起一个死循环线程不断地进行取第一个key值，如果当前时间戳大于等于该key值的socre就将它取出来进行消费删除，就可以达到延时执行的目的, 注意不需要遍历整个Zset集合，以免造成性能浪费。

Zset的排列效果如下图：

详解Redis和队列 - 文章图片

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();
        }
    }
}

实现效果如下：

详解Redis和队列 - 文章图片

以上就是详解Redis和队列的详细内容。

内容总结

以上是互联网集市为您收集整理的详解Redis和队列全部内容，希望文章能够帮你解决详解Redis和队列所遇到的程序开发问题。如果觉得互联网集市技术教程内容还不错，欢迎将互联网集市网站推荐给程序员好友。

内容备注

版权声明：本文内容由互联网用户自发贡献，该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容，请发送邮件至 gblab@vip.qq.com 举报，一经查实，本站将立刻删除。

内容手机端

扫描二维码推送至手机访问。

本文链接：https://qyyshop.com/info/588123.html

来源：【匿名】

【上一篇】redis之List操作详解【下一篇】redis可以做什么？

更多 ►

【详解Redis和队列】教程文章相关的互联网学习教程文章

使用redis的发布订阅模式实现消息队列【代码】

配置文件<?xml version="1.0" encoding="UTF-8"?> <beans xmlns="http://www.springframework.org/schema/beans"xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:aop="http://www.springframework.org/schema/aop"xmlns:tx="http://www.springframework.org/schema/tx" xmlns:util="http://www.springframework.org/schema/util"xmlns:context="http://www.springframework.org/schema/context" xmlns:mongo="http:/...

redis代替kafka做缓存队列

前言：刚上线elk一个月左右，使用的kafka作为缓存队列，但是不知道为何，中间发生过好几次，elk突然没数据了，长达好几天都没有，折腾了好久，好了，过几天又发生同样的状况。经查找，数据是有到达kafka，但是logstash读取不了。无奈之下，只能把kafka 更换为redis。 filebeat配置-------------filebeat: prospectors: - document_type: "web-hkgf-proxy-nginx-access" paths: - /data/logs/...

python 操作 redis + 消息队列使用例子【代码】

操作 redisimport redis redisPool = redis.ConnectionPool(host=‘192.168.100.50‘, port=6379, db=8) redis= redis.Redis(connection_pool=redisPool)redis.set(‘key‘,‘values‘) redis.get(‘com‘) redis.append(‘keys‘,‘values‘) redis.delete(‘keys‘)print(redis.getset(‘name‘,‘Mike‘)) #赋值name为Mike并返回上一次的value print(redis.mget([‘name‘,‘age‘])) #输出name键和age键的value print(redis...

Java利用Redis实现消息队列【代码】

应用场景为什么要用redis?二进制存储、java序列化传输、IO连接数高、连接频繁一、序列化　　这里编写了一个java序列化的工具,主要是将对象转化为byte数组,和根据byte数组反序列化成java对象; 主要是用到了ByteArrayOutputStream和ByteArrayInputStream; 注意:每个需要序列化的对象都要实现Serializable接口; 其代码如下: 1 package Utils;2 import java.io.*;3 /**4 * Created by Kinglf on 2016/10/17.5 */6 public class Objec...

[转载] 基于Redis实现分布式消息队列

转载自http://www.linuxidc.com/Linux/2015-05/117661.htm1、为什么需要消息队列？当系统中出现“生产“和“消费“的速度或稳定性等因素不一致的时候，就需要消息队列，作为抽象层，弥合双方的差异。举个例子：业务系统触发短信发送申请，但短信发送模块速度跟不上，需要将来不及处理的消息暂存一下，缓冲压力。再举个例子：调远程系统下订单成本较高，且因为网络等因素，不稳定，攒一批一起发送。再举个栗子，交互模块5：00到24...

redis适合做队列的几种方式总结

以下介绍redis的几种类型：List、Pub/Sub、Stream场景ListPub/SubStream阻塞式消费支持支持支持发布/订阅不支持支持支持重复消费不支持不支持支持持久化支持不支持支持消息堆积内存持续增长缓冲区溢出，消费者强制下线可控制队列最大长度消息会不会丢失Redis本身不保证数据完整性，存在数据丢失的可能消息积压能力Redis数据存储在内存，消息堆积对内存压力较大以上总结参考https://mp.weixin.qq.com/s/QQTG7RIOgSvSIueTqOHwww原文...

RabbitMQ与Redis队列对比【图】

本文仅针对RabbitMQ与Redis做队列应用时的情况进行对比具体采用什么方式实现，还需要取决于系统的实际需求简要介绍 RabbitMQRabbitMQ是实现AMQP（高级消息队列协议）的消息中间件的一种，最初起源于金融系统，用于在分布式系统中存储转发消息，在易用性、扩展性、高可用性等方面表现不俗。消息中间件主要用于组件之间的解耦，消息的发送者无需知道消息使用者的存在，反之亦然。Redis是一个Key-Value的NoSQL数据库，开发维护很活跃，...

JavaWeb项目架构之Redis分布式日志队列【代码】【图】

架构、分布式、日志队列，标题自己都看着唬人，其实就是一个日志收集的功能，只不过中间加了一个Redis做消息队列罢了。前言为什么需要消息队列？当系统中出现“生产“和“消费“的速度或稳定性等因素不一致的时候，就需要消息队列，作为抽象层，弥合双方的差异。比如我们系统中常见的邮件、短信发送，把这些不需要及时响应的功能写入队列，异步处理请求，减少响应时间。如何实现？成熟的JMS消息队列中间件产品市面上有很多，但是基...

java redis使用之利用jedis实现redis消息队列

应用场景最近在公司做项目，需要对聊天内容进行存储，考虑到数据库查询的IO连接数高、连接频繁的因素，决定利用缓存做。从网上了解到redis可以对所有的内容进行二进制的存储，而java是可以对所有对象进行序列化的，序列化的方法会在下面的代码中提供实现。序列化这里我编写了一个java序列化的工具，主要是对对象转换成byte[]，和根据byte[]数组反序列化成java对象；主要是用到了ByteArrayOutputStream和ByteArrayInputStream；需要...

python 结合redis 队列做一个例子【代码】

结合redis 队列做了一个例子#!/usr/bin/env python # coding: utf-8 # @Time : 2018/12/21 0021 13:57 # @Site : # @File : demos.py # @Software: PyCharm import MySQLdb import redis import json import os, time import threading from multiprocessing import Pool, Process import os, time, random import sysreload(sys) sys.setdefaultencoding(‘utf8‘)class InsertData():def __init__(self):# 去掉一些无...

REDIS - 技术教程分类

Redis 教程 Redis 简介 Redis 安装 Redis 配置 Redis 数据类型 Redis 命令 Redis 键(key) Redis 字符串(String) Redis 哈希(Hash) Redis 列表(List) Redis 集合(Set) Redis 有序集合(sorted set) Redis HyperLogLog Redis 发布订阅 Redis 事务 Redis 脚本 Redis 连接 Redis 服务器 Redis GEO Redis Stream Redis 数据备份与恢复 Redis 安全 Redis 性能测试 Redis 客户端连接 Redis 分区 Redis 全部

REDIS - 最热教程

redis中批量删除key的方法 redis设置密码，取消密码 node.js利用redis数据库缓存数据 PHP swoole和redis异步任务实现方法分析 Laravel使用Redis共享Session步骤详解 mac下卸载干净redis 如何通过命令行修改redis的配置使用Port Forwarding连接k8s集群的pod（...实现商品秒杀之——Redis Watch机制 redis雪崩、穿透、击穿、热点分别代表什...

首页 / REDIS / 详解Redis和队列

详解Redis和队列

内容导读

内容图文

概要

普通队列实现

一次生产多次消费的队列

延时队列

背景

几种延时队列

1.Java中java.util.concurrent.DelayQueue

2.Rocketmq延时队列

3.Rabbitmq延时队列（TTL+DLX实现）

Redis实现的延时消息队列适合的项目特点：

Redis实现的延时消息队列思路

内容总结

内容备注

内容手机端

【详解Redis和队列】教程文章相关的互联网学习教程文章

使用redis的发布订阅模式实现消息队列【代码】

redis代替kafka做缓存队列

python 操作 redis + 消息队列使用例子【代码】

Java利用Redis实现消息队列【代码】

[转载] 基于Redis实现分布式消息队列

redis适合做队列的几种方式总结

RabbitMQ与Redis队列对比【图】

JavaWeb项目架构之Redis分布式日志队列【代码】【图】

java redis使用之利用jedis实现redis消息队列

python 结合redis 队列做一个例子【代码】

RabbitMQ与Redis队列对比

rabbitmq和redis用作消息队列的区别

【高并发简单解决方案】redis缓存队列+mysql批量入库+php离线整合

缓存-php如何后台监控redis队列是否有值，有值执行脚本php如何后台监控redis队列是否有值，有值执行脚本

redis队列生成缓存的相关问题

REDIS - 相关标签

REDIS - 技术教程分类

REDIS - 最新教程

REDIS - 最热教程