写文章
提问题
写文章
提问题
大概我有多个python程序在运行，我希望有一个公用的计数器，不同的python程序都可以加1，也可以获取当前计数。
实现的方法很多，通过缓存或redis都可以实现。
但这里，我想要一个更轻松简易的实现方式。更方便，更快捷。
比如引入一个python模块，就立刻能解决这个问题。
有没有啥现成的解决方式，自己写一个python模块，用指定路径的文件作为记录，一来是锁的问题，二来是不如内存这么快，有没有类似简易快捷的解决方案。
jerkzhang
我感觉，还是就redis非常适合这个，可靠高效。
不过我觉得，你想要找的是“python 不同进程之间 共享内存”。
python的标准库 multiprocessing.shared_memory --- 可跨进程直接访问的共享内存
但是我感觉你要找的是基于这个做好的工具。
类似的也有一些，但你想要的应该是比较轻量级，使用起来比较方便的；nmap我考虑过的，但感觉似乎有更适合你需求的。
似乎我没注意，也许你可以考虑使用标准模块SharedMemory类自己写一个。
示例：
#coding=utf-8
import ctypes
from multiprocessing import shared_memory, Lock
import os
class SharedCounter:
def __init__(self, name=None):
self._shm = None
self._counter = None
self.lock = Lock()
self._created_shm = False
if name is None:
# 创建新的共享内存
self._shm = shared_memory.SharedMemory(create=True, size=ctypes.sizeof(ctypes.c_int64))
self._counter = ctypes.cast(
ctypes.addressof(ctypes.c_int64.from_buffer(self._shm.buf)),
ctypes.POINTER(ctypes.c_int64)
)
self._counter[0] = 0
self._created_shm = True
else:
# 连接到已有的共享内存
self._shm = shared_memory.SharedMemory(name=name)
self._counter = ctypes.cast(
ctypes.addressof(ctypes.c_int64.from_buffer(self._shm.buf)),
ctypes.POINTER(ctypes.c_int64)
)
self._created_shm = False
def __del__(self):
if self._shm is not None:
self._shm.close()
# 不要自动unlink共享内存，以便其他进程可以继续使用
@classmethod
def connect(cls, name):
instance = cls.__new__(cls)
instance.__init__(name=name)
return instance
def increment(self):
with self.lock:
self._counter[0] += 1
def get_value(self):
return self._counter[0]
def reset(self):
with self.lock:
self._counter[0] = 0
def get_shared_memory_name(self):
return self._shm.name if self._shm else None
def close(self):
if self._shm is not None:
self._shm.close()
def unlink(self):
if self._shm is not None and self._created_shm:
self._shm.unlink()
# 示例使用
if __name__ == "__main__":
# 创建计数器
counter = SharedCounter()
# 打印共享内存的名称
print(f"Shared memory name: {counter.get_shared_memory_name()}")
# 增加计数器
counter.increment()
# 获取当前值
value = counter.get_value()
print(f"Current value: {value}")
# 重置计数器
counter.reset()
print(f"After reset: {counter.get_value()}")
# 记得在最后一个进程使用后关闭共享内存
counter.close()但是还是不推荐这样自己写，还是强烈推荐使用redis来，redis非常适合做这个。