Bagaimana untuk melaksanakan baris gilir kelewatan dalam Redis

Redis melaksanakan baris gilir kelewatan

Baris gilir kelewatan Redis

Redis melaksanakan baris gilir mesej tertunda melalui set tertib (ZSet) Ya, ZSet mempunyai Skor harta yang boleh digunakan untuk menyimpan masa pelaksanaan tertunda.

Tetapi ia memerlukan gelung tak terhingga untuk menyemak tugas, yang akan menggunakan sumber sistem

class RedisDelayQueue(object):
    """Simple Queue with Redis Backend
    dq = RedisDelayQueue('delay:commtrans')
    dq.put( 5 ,{'info':'测试 5555','time': timestamp_to_datetime_str(t + 5)})
    print(dq.get())
    """
    def __init__(self, name, namespace='queue'):
        """The default connection parameters are: host='localhost', port=6379, db=0"""
        self.__db = get_redis_engine(database_name='spdb')
        self.key = '%s:%s' % (namespace, name)
    def qsize(self):
        """Return the approximate size of the queue."""
        return self.__db.zcard(self.key)
    def empty(self):
        """Return True if the queue is empty, False otherwise."""
        return self.qsize() == 0
    def rem(self, value):
        return self.__db.zrem(self.key, value)
    def get(self):
        # 获取任务，以0和当前时间为区间，返回一条在当前区间的记录
        items = self.__db.zrangebyscore(self.key, 0, int(time.time()), 0, 1)
        if items:
            item = items[0]
            if self.rem(item):  # 解决并发问题  如能删就让谁取走
                return json.loads(item)
        return None
    def put(self, interval, item):
        """:param interval 延时秒数"""
        # 以时间作为score，对任务队列按时间戳从小到大排序
        """Put item into the queue."""
        d = json.dumps(item)
        return self.__db.zadd(self.key, {d: int(time.time()) + int(interval)})

Penyelesaian pengoptimuman Redis untuk melaksanakan baris gilir kelewatan

Aplikasi baris gilir kelewatan

Terbaru Dalam projek baharu jabatan pembangunan, salah satu fungsi utama ialah tolak pintar, iaitu untuk menolak mesej peringatan yang sepadan kepada pengguna pada masa tertentu berdasarkan gelagat pengguna, seperti senario perniagaan berikut:

• Selepas pengguna mengklik pada item cas semula dan tidak mengecas semula dalam masa setengah jam, peringatan cas semula yang belum selesai akan ditolak kepada pengguna.

• Tolak peringatan teruskan membaca kepada pengguna 2 jam selepas gelagat membaca terakhir pengguna.

• Selama beberapa minit selepas pengguna baru mendaftar atau keluar dari apl, tolak mesej pengesyoran yang sesuai kepada pengguna.

• …

Ciri biasa senario di atas ialah menangguhkan tempoh masa tertentu selepas peristiwa dicetuskan sebelum melaksanakan sesuatu tertentu tugasan. Jika peristiwa mencetuskan masa Ia boleh dilihat bahawa logik di atas juga boleh bersamaan dengan melaksanakan tugas tertentu pada titik masa tertentu (titik masa pencetus peristiwa + tempoh masa tunda).

Baris gilir kelewatan biasanya digunakan untuk mencapai keperluan seperti ini Mesej kelewatan yang dibuat perlu mengandungi maklumat seperti masa kelewatan tugasan atau titik masa pelaksanaan tugasan Apabila tugasan memenuhi syarat masa dan perlu dilaksanakan. mesej itu akan menjadi Penggunaan, iaitu, anda boleh menentukan titik masa di mana mesej dalam baris gilir digunakan.

Pelaksanaan baris gilir kelewatan

Dalam persekitaran yang berdiri sendiri, JDK sudah disertakan dengan banyak komponen yang boleh melaksanakan fungsi baris gilir kelewatan, seperti DelayQueue, Timer, ScheduledExecutorService dan komponen lain boleh membuat tugas tertunda dengan mudah, tetapi penggunaan komponen di atas secara amnya memerlukan tugasan disimpan dalam ingatan Terdapat risiko kehilangan tugas apabila perkhidmatan dimulakan semula, dan saiz tugasan dihadkan oleh memori. yang juga mengakibatkan penggunaan ingatan jangka panjang , tidak fleksibel dan biasanya sesuai untuk program atau projek klien satu proses dengan keperluan tugas yang rendah.

Dalam persekitaran yang diedarkan, baris gilir kelewatan tidak boleh dilaksanakan dengan pasti dan cekap menggunakan komponen JDK sendiri lazimnya perlu memperkenalkan perisian tengah atau rangka kerja pihak ketiga.

Contohnya, rangka kerja penjadualan tugas klasik biasa Quartz atau rangka kerja lain berdasarkan rangka kerja ini xxl-job Fungsi utama rangka kerja ini adalah untuk melaksanakan tugas berjadual atau tugas berkala 🎜> Sebelum ia digunakan secara meluas, fungsi biasa seperti membatalkan pesanan kerana tamat masa dan tidak membayar telah dilaksanakan oleh tugasan berjadual telah digunakan untuk menentukan sama ada titik masa untuk mencetuskan pelaksanaan telah dicapai. RedisRabbitMQWalau bagaimanapun, memandangkan tugas berjadual memerlukan tempoh tertentu, selang antara imbasan berkala sukar dikawal Jika terlalu pendek, ia akan menyebabkan banyak imbasan tidak bermakna dan meningkatkan tekanan sistem Jika terlalu lama , ia akan menyebabkan ralat masa pelaksanaan yang berlebihan Besar, dan boleh menyebabkan bilangan rekod bertindan yang diproses dalam satu imbasan menjadi terlalu besar.

Selain itu, ia juga merupakan cara biasa untuk menggunakan

sebagai baris gilir kelewatan Contohnya, melalui

dan baris gilir surat mati untuk mencapai penghantaran mesej yang tertunda, mengambil. mengambil kira MQ Mesej yang dihantar tidak boleh dipadam atau diubah suai dengan mudah, iaitu tugasan tidak boleh dibatalkan atau titik masa pelaksanaan tugasan boleh ditukar Pada masa yang sama, mesej tidak boleh dinyahduplikasi dengan mudah untuk melaksanakan barisan kelewatan dalam projek. Struktur data RabbitMQ daripada TTLMQMQ juga boleh mencapai kesan baris gilir kelewatan, dan lebih fleksibel Ia dapat merealisasikan beberapa ciri yang

tidak dapat dilakukan, jadi projek akhirnya menggunakan

untuk mencapai baris gilir kelewatan, dan mengoptimumkan serta merangkumnya. Prinsip pelaksanaan Rediszset ialah menggunakan atribut MQ Redis akan mengisih elemen dalam koleksi

daripada kecil kepada besar mengikut

, dan menggunakan <.> perintah kepada zset Tambah elemen dalam , seperti yang ditunjukkan dalam arahan berikut, di mana nilai score ialah mesej tugas tertunda, format mesej boleh ditakrifkan mengikut perniagaan dan nilai redis ialah masa titik pelaksanaan tugas, seperti cap waktu milisaat 13 digit. Selepas tugasan zset

zadd delayqueue 1614608094000 taskinfo

score ditambah, logik untuk mendapatkan tugasan hanya perlu menapis elemen yang nilai zaddnya kurang daripada cap masa semasa daripada zset Hasilnya ialah tugasan yang perlu dilaksanakan pada nod masa semasa melalui perintah value untuk mendapatkan, seperti yang ditunjukkan dalam arahan berikut, dengan score ialah cap masa semasa Anda boleh menggunakan

untuk mengehadkan bilangan rekod yang ditarik setiap kali untuk menghalang nombor daripada rekod yang diperoleh pada satu masa daripada terlalu besar.

zrangebyscore delayqueue 0 timestamp limit 0 1000

在实际实现过程中，从zset中获取到当前需要执行的任务后，需要先确保将任务对应的元素从zset中删除，删除成功后才允许执行任务逻辑，这样是为了在分布式环境下，当存在多个线程获取到同一任务后，利用redis删除操作的原子性，确保只有一个线程能够删除成功并执行任务，防止重复执行。

实际任务的执行通常会再将其发送至MQ异步处理，将“获取任务”与“执行任务”两者分离解耦，更加灵活，“获取任务”只负责拿到当前时间需要执行的任务，并不真正运行任务业务逻辑，因此只需相对少量的执行线程即可，而实际的任务执行逻辑则由MQ消费者承担，方便调控负载能力。

整体过程如下图所示。

采用zset做延时队列的另一个好处是可以实现任务的取消和任务执行时间点的更改，只需要将任务信息从zset中删除，便可取消任务，同时由于zset拥有集合去重的特性，只需再次写入同一个任务信息，但是value值设置为不同的执行时间点，便可更改任务执行时间，实现单个任务执行时间的动态调整。

了解实现原理后，再进行具体编程实现。创建延时任务较为简便，准备好任务消息和执行时间点，写入zset即可。获取延时任务最简单的方案是通过定时任务，周期性地执行上述逻辑，如下代码所示。

@XxlScheduled(cron = "0/5 * * * * ?", name = "scan business1 delayqueue")
public void scanBusiness1() {
	// 某业务逻辑的zset延迟队列对应的key
	String zsetKey = "delayqueue:business1";
	while (true) {
		// 筛选score值小于当前时间戳的元素，一次最多拉取1000条
		Set<String> tasks = stringRedisTemplate.opsForZSet().rangeByScore(zsetKey, 0, System.currentTimeMillis(), 0, 1000);
		if (CollectionUtils.isEmpty(tasks)) {
			// 当前时间下已没有需要执行的任务，结束本次扫描
			return;
		}
		for (String task : tasks) {
			// 先删除，再执行，确保多线程环境下执行的唯一性
			Boolean delete = stringRedisTemplate.delete(task);
			if (delete) {
				// 删除成功后，将其再发送到指定MQ异步处理，将“获取任务”与“执行任务”分离解耦
				rabbitTemplate.convertAndSend("exchange_business1", "routekey_business1", task);
			}
		}
	}
}

上述方案使用xxl-job做分布式定时任务，间隔5秒执行一次，代码借助spring提供的api来完成redis和MQ的操作。

由于是分布式定时任务，每次执行只有一个线程在获取任务，机器利用率低，当数据规模较大时，单靠一个线程无法满足吞吐量要求，因此这种方案只适用于小规模数据量级别。

此处间隔时间也可适当调整，例如缩短为1秒，调整所需考虑原则在上文已提到：间隔太短会造成很多无意义的扫描，且增大系统压力，太长又会造成执行时间误差太大。

为了提升整体吞吐量，考虑不使用分布式定时任务，对集群内每台机器（或实例）均设置独立的定时任务，同时采用多个zset队列，以数字后缀区分。

假设有M个zset队列，创建延时消息时选取消息的某个ID字段，计算hash值再对M取余，根据余数决定发送到对应数字后缀的zset队列中（分散消息，此处ID字段选取需要考虑做到均匀分布，不要造成数据倾斜）。

队列数量M的选取需要考虑机器数量N，理想情况下有多少台机器就定义多少个队列，保持M与N基本相等即可。

因为队列太少，会造成机器对队列的竞争访问处理，队列太多又会导致任务得不到及时的处理。

为了灵活应对集群机器数量的变化，建议将队列数量配置为动态可调的，例如采用分布式配置中心的方案。

每台机器在触发定时任务时，需要通过适当的负载均衡来决定从哪个队列拉取消息，负载均衡的好坏也会影响整个集群的效率，如果负载分布不均可能会导致多台机器竞争处理同一队列，降低效率。

一个简单实用的做法是利用redis的自增操作再对队列数量取余即可，只要保持队列数量和机器数量基本相等，这种做法在很大程度上就可以保证不会有多台机器竞争同一队列。

至于每台机器从对应zset中的任务获取逻辑，仍然和前面代码一致。以上方式简化实现代码如下所示。

@Scheduled(cron = "0/5 * * * * ?")
public void scanBusiness1() {
	// 队列数量M，考虑动态配置，保持和机器数量基本一致
	int M = 10;
	// redis自增key，用于负载均衡
	String incrKey = "incrkey:delayqueue:business1";
	// 每台机器执行时，从不同的zset中拉取消息，尽量确保不同机器访问不同zset
	String zsetKey = "delayqueue:business1:" + (stringRedisTemplate.opsForValue().increment(incrKey) % M);
	while (true) {
		// 此处逻辑和前面代码一致，省略。。。
	}
}

上述方案和第一种方案的主要的不同点在于zsetKey的获取上，这里是根据负载均衡算法算出来的，确保每台机器访问不同zset并拉取消息，同时定时任务采用spring提供的进程内注解@Scheduled，集群内每台机器都会间隔5秒执行，因此相比之前的方案，能够较为明显地提升整个集群的吞吐量。

但是这种方案的步骤相对更为复杂，需要动态配置队列数量，同时在创建延时任务时需要选择合适的消息ID字段来决定发送的目标zset队列，此处还要考虑均匀分布，整体实现要考虑的因素较多。

上面一种方案已经能够较好地满足整体吞吐量要求，但其缺点是步骤相对复杂，因此项目中没有采用这种方案，而是采用下面一种也能满足吞吐量要求，步骤相对简单，又方便通用化的方案。

该方案不使用定时任务，而是单独启动后台线程，在线程中执行永久循环，每次循环逻辑为：从目标zset中获取score值小于当前时间戳的元素集合中的score最小的那个元素，相当于获取当前时间点需要执行且执行时间点最早的那个任务，如果获取不到，表示当前时间点下暂无需要执行的任务，则线程休眠100ms（可视情况调整），否则，对获取到的元素进行处理，在分布式多线程环境下，仍然需要先删除成功才能进行处理。

此外，考虑到每个线程获取元素后都需要再次访问redis尝试删除操作，为了避免多线程争抢浪费资源，降低效率，这里采用lua脚本将获取和删除操作原子化。lua脚本逻辑代码如下所示。

local zsetKey = 'delayqueue'
local timestamp = 1614608094000
local items = redis.call('zrangebyscore',zsetKey,0,timestamp,'limit',0,1)
if #items == 0 then
    return ''
else
    redis.call('zremrangebyrank',zsetKey,0,0)
    return items[1]
end

其中timestamp为当前时间戳，通过在zrangebyscore命令中指定limit为1来获取score最小的元素，若获取不到，即结果集长度为0，则返回空字符串，否则，通过zremrangebyrank命令删除头部元素，即score最小的元素，也就是之前获取到的那个元素，由于redis内部保证lua脚本的原子性，上述获取并删除的操作能够运行无误。具体JAVA实现中还对其进行了多线程操作的封装和通用化的抽象，使不同业务都能够使用该组件实现延时队列。具体实现代码如下所示。

/**
 * 基于ZSET实现消息延迟处理，score存储执行时间点，到达时间点即会向指定队列发送该消息；
 * 定义一个继承本类的bean即可；
 */
public abstract class AbstractDelayedMsgScanTrigger implements Runnable, DisposableBean {
	private static final RedisScript<String> TRY_GET_AND_DEL_SCRIPT;
	static {
		// 获取并删除的lua脚本，使用spring提供的api
		String sb = "local items = redis.call(&#39;zrangebyscore&#39;,KEYS[1],0,ARGV[1],&#39;limit&#39;,0,1)\n" +
				"if #items == 0 then\n" +
				"\treturn &#39;&#39;\n" +
				"else\n" +
				"\tredis.call(&#39;zremrangebyrank&#39;,KEYS[1],0,0)\n" +
				"\treturn items[1]\n" +
				"end";
		// 自有工具类，只要能创建出spring包下的 RedisScript 的实现类对象均可
		TRY_GET_AND_DEL_SCRIPT = RedisScriptHelper.createScript(sb, String.class);
	}
	private final ThreadPoolExecutor EXECUTOR = new ThreadPoolExecutor(getThreadNum(), getThreadNum(),
			0, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<>(), new NamedThreadFactory(getThreadNamePrefix()));
	private volatile boolean quit = false;
	@Autowired
	private StringRedisTemplate stringRedisTemplate;
	@Autowired
	private RabbitTemplate rabbitTemplate;
	@PostConstruct
	public void startScan() {
		// bean构建完成后，启动若干执行线程
		int threadNum = getThreadNum();
		for (int i = 0; i < threadNum; i++) {
			EXECUTOR.execute(this);
		}
	}
	@Override
	public void run() {
		while (!quit) {
			try {
				// 循环，采用lua获取当前需要执行的任务并将其从redis中删除
				String msg = stringRedisTemplate.execute(TRY_GET_AND_DEL_SCRIPT,
						Lists.newArrayList(getDelayedMsgSourceKey()), String.valueOf(System.currentTimeMillis()));
				if (StringUtils.isNotBlank(msg)) {
					// 消息不为空，表示获取任务成功，将其再发送到指定MQ异步处理，将“获取任务”与“执行任务”分离解耦
					rabbitTemplate.convertAndSend(getSendExchange(), getSendRoutingKey(), msg);
				} else {
					// 获取不到任务，表示当前时间点下暂无需要执行的任务，则线程休眠1S（可视情况调整）
					SleepUtils.sleepSeconds(1);
				}
			} catch (Exception e) {
				Logs.MSG.error("delayed msg scan error, sourceKey:{}", getDelayedMsgSourceKey(), e);
			}
		}
	}
	@Override
	public void destroy() throws Exception {
		quit = true;
	}
	public void setQuit(boolean quit) {
		this.quit = quit;
	}
	/**
	 * 获取消息的工作线程数量
	 */
	protected abstract int getThreadNum();
	/**
	 * 线程名称前缀，方便问题定位
	 */
	protected abstract String getThreadNamePrefix();
	/**
	 * 存放延迟消息的ZSET队列名
	 */
	protected abstract String getDelayedMsgSourceKey();
	/**
	 * 消息到达执行时间点时将其通过指定 exchange 发送到实时消费队列中
	 */
	protected abstract String getSendExchange();
	/**
	 * 消息到达执行时间点时将其通过指定 routingKey 发送到实时消费队列中
	 */
	protected abstract String getSendRoutingKey();
}

在具体业务应用中，只需定义一个继承上述类的bean即可，需要实现的方法主要是提供一些配置，比如该业务对应的zset延时队列名称，同时工作拉取消息的线程数量，由于采用rabbitMq，因此这里需要提供exchange和routingKey。

实际使用中只需向该zset队列中添加消息，并将score设为该任务需要执行的时间点（此处为13位毫秒时间戳），则到该时间点后，上述组件便会将该消息从zset中取出并删除，再将其通过指定的路由发送到实时MQ消费队列中，由消费者负责执行任务业务逻辑。目前该组件在项目中正常平稳运行。

注意：

本文结合项目中的实际需求介绍了延时队列的应用场景，分析了延时队列的多种实现，重点讲述了利用redis实现延时队列的原理，对其实现方案进行比较与优化，并将最终方案实际运用于项目需求中。

Atas ialah kandungan terperinci Bagaimana untuk melaksanakan baris gilir kelewatan dalam Redis. Untuk maklumat lanjut, sila ikut artikel berkaitan lain di laman web China PHP!