Python中的队列PriorityQueue优先级队列 三

比较两个对象大小

a,b这两个对象是不支持 >大于这个比较操作的

class T : passa =T()b = T()print( a > b )

Traceback (most recent call last):  File 'D:\zsoft\project\eclipse_python\pysimple_ui\otehr\marshal_usage.py', line 225, in <module>    print( a > b )TypeError: '>' not supported between instances of 'T' and 'T'

代码稍微改造一下，支持大于>比较操作 __gt__函数实现自己的逻辑

class T : def __gt__(self, other): return id(self) > id(other) passa =T()b = T()print( a > b )

实现对象比较的几个方法

• __cmp__() 比较
• __gt__() 大于 >
• __lt__()， 小于 <

我们知道Python内置的这些方法其实是用作对象比较的

借用网上的代码基于2.x版本，我拿过来改造了一下。如下。

class Door(object):    def __init__(self):        self.value = 0     def __cmp__(self, other):        print ('====my cmp====')        if self.value > other.value:            return 1        if self.value < other.value:            return -1        return 0     def __gt__(self, other):        print ('====my gt====')        return self.value > other.value     def __lt__(self, other):        print ('====my lt====')        return self.value < other.value a = Door()b = Door() a.value = 20b.value = 20 print ( a == b )print ( a > b)print ( a < b)

优先使用

• __gt__(), 对应 >
• __lt__(), 对应 <
• __eq__()， 对应 =
• 如果找不到则使用 __cmp__()

最牛 functools.total_ordering()

def total_ordering(cls): '''Class decorator that fills in missing ordering methods''' convert = { '__lt__': [('__gt__', lambda self, other: not (self < other or self == other)), ('__le__', lambda self, other: self < other or self == other), ('__ge__', lambda self, other: not self < other)], '__le__': [('__ge__', lambda self, other: not self <= other or self == other), ('__lt__', lambda self, other: self <= other and not self == other), ('__gt__', lambda self, other: not self <= other)], '__gt__': [('__lt__', lambda self, other: not (self > other or self == other)), ('__ge__', lambda self, other: self > other or self == other), ('__le__', lambda self, other: not self > other)], '__ge__': [('__le__', lambda self, other: (not self >= other) or self == other), ('__gt__', lambda self, other: self >= other and not self == other), ('__lt__', lambda self, other: not self >= other)] } roots = set(dir(cls)) & set(convert) if not roots: raise ValueError('must define at least one ordering operation: < > <= >=') root = max(roots) # prefer __lt__ to __le__ to __gt__ to __ge__ for opname, opfunc in convert[root]: if opname not in roots: opfunc.__name__ = opname opfunc.__doc__ = getattr(int, opname).__doc__ setattr(cls, opname, opfunc) return cls

如果是实现了__lt__()，那么程序会自动帮你实现了__gt__(), __ge__()等等的N多函数

使用functools.total_ordering()装饰器定义一个队列任务元素

@functools.total_orderingclass Task():	def __init__(self, customer, description, entry_counter):		self.customer = customer		self.description = description		self.priority = customer.get_signup_year()		self.entry_counter = entry_counter		def __repr__(self):		return 'Task('   str((self.priority, self.entry_counter, self.description, self.customer))   ')'		def __lt__(self, other):		return (self.priority, self.entry_counter) < (other.priority, other.entry_counter)	def __eq__(self, other):		return (self.priority, self.entry_counter) == (other.priority, other.entry_counter)

接着定义一个数据类Customer

@dataclassclass Customer(): signup_date: date first_name: str last_name: str def get_signup_year(self): return self.signup_date.year def __repr__(self): return str((self.signup_date.strftime('%Y-%m-%d'), self.first_name ' ' self.last_name))

最后使用优先级队列实现生产者消费者

import randomimport functoolsfrom datetime import datefrom dataclasses import dataclassfrom queue import PriorityQueuedef generate_customers(n):	first_names = ['Jessica', 'Mary', 'Patricia', 'Jennifer', 'Elizabeth', 'James', 'John', 'Robert', 'Michael', 'Noah', 'Damian']	last_names = ['Smith', 'Johnson', 'Williams', 'Brown', 'Davis', 'Rodriguez', 'Murphy', 'Walsh', 'Wilson', 'Taylor']	customers = []	for i in range(n):		c = Customer(			first_name=random.choice(first_names),			last_name=random.choice(last_names),			signup_date = date(				year=random.randint(2020, 2022),				month=random.randint(1, 12),				day=random.randint(1, 28)			)		)		customers.append(c)	return customersif __name__ == '__main__':	random.seed(3)	customers = generate_customers(5)	q = PriorityQueue()	print('Customers arriving...')	entry_counter = 0	for customer in customers:		task = Task(				customer=customer,				description=random.choice(['Delivery delay', 'Poor product quality', 'Unavailable products']),				entry_counter=entry_counter			)			q.put(task)				entry_counter  = 1		print(task) 	print()	print('Customers served...')	while not q.empty():		next_customer = q.get()		print(next_customer)

完整的代码

实际上这个优先级是自己定义的。

import randomimport functoolsfrom datetime import datefrom dataclasses import dataclassfrom queue import PriorityQueue@dataclassclass Customer(): signup_date: date first_name: str last_name: str def get_signup_year(self): return self.signup_date.year def __repr__(self): return str((self.signup_date.strftime('%Y-%m-%d'), self.first_name ' ' self.last_name))def generate_customers(n): first_names = ['Jessica', 'Mary', 'Patricia', 'Jennifer', 'Elizabeth', 'James', 'John', 'Robert', 'Michael', 'Noah', 'Damian'] last_names = ['Smith', 'Johnson', 'Williams', 'Brown', 'Davis', 'Rodriguez', 'Murphy', 'Walsh', 'Wilson', 'Taylor'] customers = [] for i in range(n): c = Customer( first_name=random.choice(first_names), last_name=random.choice(last_names), signup_date = date( year=random.randint(2020, 2022), month=random.randint(1, 12), day=random.randint(1, 28) ) ) customers.append(c) return [e-mail]_orderingclass Task(): def __init__(self, customer, description, entry_counter): self.customer = customer self.description = description self.priority = customer.get_signup_year() self.entry_counter = entry_counter def __repr__(self): return 'Task(' str((self.priority, self.entry_counter, self.description, self.customer)) ')' def __lt__(self, other): return (self.priority, self.entry_counter) < (other.priority, other.entry_counter) def __eq__(self, other): return (self.priority, self.entry_counter) == (other.priority, other.entry_counter)if __name__ == '__main__': random.seed(3) customers = generate_customers(5) q = PriorityQueue() print('Customers arriving...') entry_counter = 0 for customer in customers: task = Task( customer=customer, description=random.choice(['Delivery delay', 'Poor product quality', 'Unavailable products']), entry_counter=entry_counter ) q.put(task) entry_counter = 1 print(task) print() print('Customers served...') while not q.empty(): next_customer = q.get() print(next_customer)