In the main thread, a database (SQLite) is created on the peewee .
After that, additional QThread threads are QThread .
Additional streams collect information and send it to the main stream via a signal.
In the main thread I tried QueuedConnection and BlockingQueuedConnection for signal processing.

But it still happens that 2-3 additional threads simultaneously cause one signal.
As a result, the database gets up and the work stops.

How can I fix it?

    1 answer 1

    It is a pity that you do not publish a minimal example to demonstrate the problem.

    Here is a small imitation of what you said. DB replaced by QListWidget. Try it.

     import sys import threading from PyQt5.QtCore import * from PyQt5.QtGui import * from PyQt5.QtWidgets import * def thread(my_func): """ Запускает функцию в отдельном потоке """ def wrapper(*args, **kwargs): my_thread = threading.Thread(target=my_func, args=args, kwargs=kwargs) my_thread.start() return wrapper @thread def processing(signal): """ Эмулирует обработку каких-то данных """ res = [i for i in 'hello'] QThread.msleep(10) signal.emit(res) # Посылаем сигнал в котором передаём полученные данные class MyWidget(QWidget): my_signal = pyqtSignal(list, name='my_signal') def __init__(self, parent=None): super(MyWidget, self).__init__(parent) self.n_n = 1 self.button = QPushButton("Emit your signal!", self) self.listWidget = QListWidget(self) labels = ("Item 1", "Item 2", "Item 3") self.listWidget.addItems(labels) self.mainLayout = QHBoxLayout() self.setLayout(self.mainLayout) self.mainLayout.addWidget(self.button) self.mainLayout.addWidget(self.listWidget) # При нажатии на кнопку запускаем обработку данных self.button.clicked.connect(lambda: processing(self.my_signal)) # Обработчик сигнала self.my_signal.connect(self.mySignalHandler, Qt.QueuedConnection) def mySignalHandler(self, data): # Вызывается для обработки сигнала #print(data) item = QListWidgetItem() itemNumber = 0 # добавляем в верхнюю строку item.setText("{} {}".format(self.n_n, "".join(data))) self.listWidget.insertItem(itemNumber, item) self.n_n += 1 def testProcessing(self): """ для автоматического тестирования после запуска приложения """ for i in range(1, 6): processing(self.my_signal) if __name__ == '__main__': app = QApplication(sys.argv) window = MyWidget() window.show() # Два тестовых прогона: первый сразу, второй через 2 секунды. window.testProcessing() QTimer.singleShot(2000, window.testProcessing) app.exec_()

    enter image description here

    • Only I use QThread - GoldenEagle
    • @GoldenEagle publish a minimal example to demonstrate the problem. - S. Nick