设置队列元素最大个数

Administrator

2024-11-22 1b3d994efcba0236537f3aac3088658bcdbfd761

设置队列元素最大个数

 db/redis_manager.py

@@ -31,7 +31,7 @@


class RedisUtils:
    __async_task_queue = queue.Queue()
    __async_task_queue = queue.Queue(maxsize=1024)

    @classmethod
    def exec(cls, method_name, key, lamada_method):

 db/redis_manager_delegate.py

@@ -33,7 +33,7 @@


class RedisUtils:
    __async_task_queue = queue.Queue()
    __async_task_queue = queue.Queue(maxsize=1000)

    @classmethod
    def exec(cls, method_name, key, lamada_method):

 huaxin_client/l2_client.py

@@ -50,7 +50,7 @@
SZ_Securities = [b"002456", b"002849", b"002281", b"002336", b"000936", b"000920", b"000757", b"002896", b"002725",
                 b"000952", b"000526", b"000753", b"000681", b"002088", b"002436"]
SZ_Bond_Securities = [b"100303", b"109559", b"112617"]
set_codes_data_queue = queue.Queue()
set_codes_data_queue = queue.Queue(maxsize=1000)
market_code_dict = {}

ENABLE_NGST = True

 huaxin_client/l2_data_manager.py

@@ -28,7 +28,7 @@
tmep_transaction_queue_dict = {}
target_codes = set()
target_codes_add_time = {}
common_queue = queue.Queue()
common_queue = queue.Queue(maxsize=1000)


# L2上传数据管理器
@@ -161,7 +161,7 @@
        if code not in self.temp_transaction_queue_dict:
            self.temp_transaction_queue_dict[code] = collections.deque()
        if code not in self.temp_log_queue_dict:
            self.temp_log_queue_dict[code] = queue.Queue()
            self.temp_log_queue_dict[code] = queue.Queue(maxsize=1000)
        if code not in self.upload_l2_data_task_dict:
            t1 = threading.Thread(target=lambda: self.__run_upload_order_task(code), daemon=True)
            t1.start()

 huaxin_client/l2_market_client.py

@@ -23,7 +23,7 @@
Multicast_Address2 = "udp://224.224.224.234:7890"
Local_Interface_Address = constant.LOCAL_IP

set_codes_data_queue = queue.Queue()
set_codes_data_queue = queue.Queue(maxsize=1000)
market_code_dict = {}



 l2/huaxin/huaxin_target_codes_manager.py

@@ -16,7 +16,7 @@
from utils import tool, global_util, init_data_util

redisManager = redis_manager.RedisManager(4)
l2_codes_queue = queue.Queue()
l2_codes_queue = queue.Queue(maxsize=1000)


# 华鑫Level2订阅代码管理

 main.py

@@ -80,16 +80,16 @@
        log.close_print()

        # L2读其他写
        queue_other_w_l2_r = multiprocessing.Queue()
        queue_other_w_l2_r = multiprocessing.Queue(maxsize=1000)
        # l1
        queue_l1_w_strategy_r = multiprocessing.Queue()
        queue_l1_r_strategy_w = multiprocessing.Queue()
        queue_l1_w_strategy_r = multiprocessing.Queue(maxsize=1000)
        queue_l1_r_strategy_w = multiprocessing.Queue(maxsize=1000)

        # 交易读策略写
        queue_strategy_w_trade_r = multiprocessing.Queue()
        queue_strategy_w_trade_r_for_read = multiprocessing.Queue()
        queue_strategy_w_trade_r = multiprocessing.Queue(maxsize=1000)
        queue_strategy_w_trade_r_for_read = multiprocessing.Queue(maxsize=1000)
        # 策略读交易写
        queue_strategy_r_trade_w = multiprocessing.Queue()
        queue_strategy_r_trade_w = multiprocessing.Queue(maxsize=1000)

        # 下单,撤单ipc地址
        order_ipc_addr, cancel_order_ipc_addr = "ipc://trade_order.ipc", "ipc://trade_cancel_order.ipc"

 output/kp_client_msg_manager.py

@@ -12,7 +12,7 @@

CLIENT_IDS = ["zjb", "hxh"]

__temp_msg_queue = queue.Queue()
__temp_msg_queue = queue.Queue(maxsize=1000)


class MsgQueueManager:
@@ -21,7 +21,7 @@
    # 添加消息,2s内有效
    def add_msg(self, client_id, msg):
        if client_id not in self.__queue_dict:
            self.__queue_dict[client_id] = queue.Queue()
            self.__queue_dict[client_id] = queue.Queue(maxsize=1000)

        self.__queue_dict[client_id].put_nowait((time.time() + 2, msg))


 servers/huaxin_trade_server.py

@@ -57,7 +57,7 @@
from trade.trade_manager import CodesTradeStateManager
from utils import socket_util, middle_api_protocol, tool, huaxin_util, global_util, trade_util, init_data_util

trade_data_request_queue = queue.Queue()
trade_data_request_queue = queue.Queue(maxsize=1000)


class MyTCPServer(socketserver.TCPServer):

 trade/huaxin/huaxin_trade_api.py

@@ -25,7 +25,7 @@
from utils import socket_util, huaxin_util, tool

__response_thread_pool = concurrent.futures.ThreadPoolExecutor(max_workers=15)
__save_data_queue = queue.Queue()
__save_data_queue = queue.Queue(maxsize=1000)


def __run_save_data():

 trade/huaxin/huaxin_trade_data_update.py

@@ -18,7 +18,7 @@
from utils import huaxin_util
import concurrent.futures

trade_data_request_queue = queue.Queue()
trade_data_request_queue = queue.Queue(maxsize=1000)
__process_thread_pool = concurrent.futures.ThreadPoolExecutor(max_workers=10)