redis多线程操作

今天更新一个redis多线程操作， 可直接搬运

import redis, os, threading, queue
import pandas as pd

# 创建一个任务队列
task_queue = queue.Queue()


def read_excel(folder_path):
	total_list = []
	for filepath, dirnames, filenames in os.walk(folder_path):
		for filename in filenames:
			file_path = os.path.join(filepath, filename)
			df_total = pd.read_excel(file_path)
			list_df = df_total['product'].values.tolist()
			total_list.extend(list_df)
	print(total_list)
	print(len(total_list))
	result_list = []
	for t_pro in total_list:
		t_pro_dict = {t_pro: '20230907'}
		result_list.append(t_pro_dict)
	# # 写入 redis
	# redis_obj = RedisClass('ahrefs_filter', 9)
	# for t_pro in total_list:
	# 	t_pro_dict = {t_pro: '20230907'}
	# 	print(t_pro_dict)
	# 	redis_obj.insert_redis(t_pro_dict)
	return result_list



class RedisClass:
	def __init__(self, db_key, db_index, db_host='*.*.*.*', db_port=6379, db_password='password', filter_start_index=0, filter_end_index=0):
		# 传入DB表名，和DB序号
		self.db_key = db_key
		self.db_index = db_index
		self.db_host = db_host
		self.db_port = db_port
		self.db_password = db_password
		self.filter_start_index = filter_start_index
		self.filter_end_index = filter_end_index

		self.redis_pool = redis.ConnectionPool(host=self.db_host, port=self.db_port, password=self.db_password,
											   db=self.db_index)
		self.redis_conn = redis.Redis(connection_pool=self.redis_pool)

	def count_redis_data(self):
		# 计数： 获取redis中数据数量
		return self.redis_conn.zcard(self.db_key)

	def read_redis(self):
		# 读取redis中全部数据
		if self.filter_start_index == 0 and self.filter_end_index == 0:
			# 如果无输入查询数量， 则全表查询
			self.filter_end_index = self.redis_conn.zcard(self.db_key)
		print('查询到的数量为： {}'.format(self.filter_end_index))
		res_list = self.redis_conn.zrange(self.db_key, self.filter_start_index, self.filter_end_index)

		return [res.decode('utf-8') for res in res_list]

	def read_redis_by_score(self, zset_score):
		# 读取redis中全部数据
		res_list = self.redis_conn.zrangebyscore(self.db_key, zset_score, zset_score)

		return [res.decode('utf-8') for res in res_list]

	def insert_redis(self, redis_dict):
		flag = False
		self.redis_conn.zadd(self.db_key, redis_dict)
		return flag


# 生产者线程类
class ProducerThread(threading.Thread):
	def __init__(self, mysql_pro_info):
		super().__init__()
		self.mysql_pro_info = mysql_pro_info

	def run(self):
		for item in self.mysql_pro_info:
			task_queue.put(item)
			print(f"Produced by {self.name}: {item}")

class ConsumerThread(threading.Thread):
	def run(self):
		redis_obj = RedisClass('ahrefs_filter', 9)
		while True:
			item = task_queue.get()
			print(item)
			redis_obj.insert_redis(item)


if __name__ == '__main__':
	# 1- 读取EXCEL中的数据, 存入redis
	folder_path = r'C:\Users\admin\Desktop\0905型号'
	total_list = read_excel(folder_path)

	producer_thread = ProducerThread(total_list)
	producer_thread.start()

	for i in range(100):  # 创建100个消费者线程
		consumer_thread = ConsumerThread()
		consumer_thread.start()

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
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104