网站开发 运维 招投标,公司logo设计价格,wordpress dede,徐州建站服务提示#xff1a;文章写完后#xff0c;目录可以自动生成#xff0c;如何生成可参考右边的帮助文档 文章目录 前言一、can周期任务类构建总结 前言
提示#xff1a;这里可以添加本文要记录的大概内容#xff1a;
最近想着有时间实现总线报文收发的动态的配置#xff0c;… 提示文章写完后目录可以自动生成如何生成可参考右边的帮助文档 文章目录 前言一、can周期任务类构建总结 前言
提示这里可以添加本文要记录的大概内容
最近想着有时间实现总线报文收发的动态的配置今天特记录一下报文周期任务的动态的创建和移除 提示以下是本篇文章正文内容下面案例可供参考
一、can周期任务类构建
代码如下示例
import threading
import timeclass CycleTask():def __init__(self):self.__task_dict dict() # 周期任务的标志位字典self.__task_queue dict() # 周期任务需发送的数据字典{cycle_time:{msg_obj:data}}self.__lock_dict dict() # 每个线程对应的线程锁def create_task(self, cycle_time):功能创建周期任务参数1周期时间self.__create_task_thread(cycle_time)def __create_task_thread(self, cycle_time):功能创建周期线程参数1事件task threading.Thread(targetself.__task, args(cycle_time,))task.start()def __task(self, cycle_time):功能周期任务函数参数1周期事件lock threading.Lock()self.__lock_dict[cycle_time] locktask_flag Trueself.__task_dict[cycle_time] task_flagwhile self.__task_dict[cycle_time]:self.__lock_dict[cycle_time].acquire() # 线程上锁for msg, data in self.__task_queue[cycle_time].items():passself.__lock_dict[cycle_time].release() # 线程下锁time.sleep(cycle_time)def cycle_send_data(self, cycle_time, msg, data):功能将报文进行周期发送参数1周期事件参数2报文对象参数3数据列表if cycle_time in self.__task_queue:self.__lock_dict[cycle_time].acquire() # 线程上锁self.__task_queue[cycle_time][msg] data # {cycle_time:{msg_obj:data}}self.__lock_dict[cycle_time].release() # 线程下锁else:self.__task_queue[cycle_time] dict()self.__task_queue[cycle_time][msg] datadef cycle_msg_stop(self, cycle_time, msg):功能停止报文周期发送参数1周期事件参数2报文对象self.__lock_dict[cycle_time].acquire() # 线程上锁del self.__task_queue[cycle_time][msg]self.__lock_dict[cycle_time].release() # 线程下锁if not self.__task_queue: # 当周期任务中没有self.__task_dict[cycle_time] False 总结
我是一名车载集成测试开发工程师希望能和志同道合的朋友一起相互学习进步
