Я не нашел никакого решения для моей проблемы. Мне нужно создать приложение Python с двумя потоками, каждый из которых подключен к маршрутизатору WAMP с использованием библиотеки автобанов.
Следуйте за мной, я пишу код своего эксперимента:
wampAddress = 'ws://172.17.3.139:8181/ws'
wampRealm = 's4t'
from threading import Thread
from autobahn.twisted.wamp import ApplicationRunner
from autobahn.twisted.wamp import ApplicationSession
from twisted.internet.defer import inlineCallbacks
class AutobahnMRS(ApplicationSession):
@inlineCallbacks
def onJoin(self, details):
print("Sessio attached [Connect to WAMP Router]")
def onMessage(*args):
print args
try:
yield self.subscribe(onMessage, 'test')
print ("Subscribed to topic: test")
except Exception as e:
print("Exception:" +e)
class AutobahnIM(ApplicationSession):
@inlineCallbacks
def onJoin(self, details):
print("Sessio attached [Connect to WAMP Router]")
try:
yield self.publish('test','YOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOO')
print ("Subscribed to topic: test")
except Exception as e:
print("Exception:" +e)
class ManageRemoteSystem:
def __init__(self):
self.runner = ApplicationRunner(url= wampAddress, realm = wampRealm)
def start(self):
self.runner.run(AutobahnMRS);
class InternalMessages:
def __init__(self):
self.runner = ApplicationRunner(url= wampAddress, realm = wampRealm)
def start(self):
self.runner.run(AutobahnIM);
#class S4tServer:
if __name__ == '__main__':
server = ManageRemoteSystem()
sendMessage = InternalMessages()
thread1 = Thread(target = server.start())
thread1.start()
thread1.join()
thread2 = Thread(target = sendMessage.start())
thread2.start()
thread2.join()
Когда я запускаю это приложение python, запускается только поток 1, а позже, когда я убиваю приложение (ctrl-c), отображаются следующие сообщения об ошибках:
Sessio attached [Connect to WAMP Router]
Subscribed to topic: test
^CTraceback (most recent call last):
File "test_pub.py", line 71, in <module>
p2 = multiprocessing.Process(target = server.start())
File "test_pub.py", line 50, in start
self.runner.run(AutobahnMRS);
File "/usr/local/lib/python2.7/dist-packages/autobahn/twisted/wamp.py", line 175, in run
reactor.run()
File "/usr/local/lib/python2.7/dist-packages/twisted/internet/base.py", line 1191, in run
self.startRunning(installSignalHandlers=installSignalHandlers)
File "/usr/local/lib/python2.7/dist-packages/twisted/internet/base.py", line 1171, in startRunning
ReactorBase.startRunning(self)
File "/usr/local/lib/python2.7/dist-packages/twisted/internet/base.py", line 683, in startRunning
raise error.ReactorNotRestartable()
twisted.internet.error.ReactorNotRestartable
Мне нужно реализовать в одном приложении, которое имеет свои функции, а также должно иметь систему для связи с маршрутизатором WAMP с библиотекой автобана Python.
Другими словами, мне нужно решение, способное взаимодействовать с маршрутизатором WAMP, но в то же время это приложение не должно быть заблокировано частью автобана (я думаю, что решение состоит в том, чтобы запустить два потока, один поток управляет некоторыми функциями и второй поток управляет частью автобана).
Со схемой, которую я предложил ранее, есть еще одна проблема, необходимость отправки сообщения, в определенной теме на маршрутизаторе WAMP, из прикладной части в «нет потока автобана», этот функционал должен вызываться через конкретную функцию без блокировки других функций.
Надеюсь, я рассказал все подробности.
Большое спасибо за любой ответ
--------------------------------РЕДАКТИРОВАТЬ----------------- ----------------
После некоторых исследований я реализовал то, что мне нужно для протокола websocket, код выглядит следующим образом:
# ----- twisted ----------
class _WebSocketClientProtocol(WebSocketClientProtocol):
def __init__(self, factory):
self.factory = factory
def onOpen(self):
#log.debug("Client connected")
self.factory.protocol_instance = self
self.factory.base_client._connected_event.set()
class _WebSocketClientFactory(WebSocketClientFactory):
def __init__(self, *args, **kwargs):
WebSocketClientFactory.__init__(self, *args, **kwargs)
self.protocol_instance = None
self.base_client = None
def buildProtocol(self, addr):
return _WebSocketClientProtocol(self)
# ------ end twisted -------
lass BaseWBClient(object):
def __init__(self, websocket_settings):
#self.settings = websocket_settings
# instance to be set by the own factory
self.factory = None
# this event will be triggered on onOpen()
self._connected_event = threading.Event()
# queue to hold not yet dispatched messages
self._send_queue = Queue.Queue()
self._reactor_thread = None
def connect(self):
log.msg("Connecting to host:port")
self.factory = _WebSocketClientFactory(
"ws://host:port",
debug=True)
self.factory.base_client = self
c = connectWS(self.factory)
self._reactor_thread = threading.Thread(target=reactor.run,
args=(False,))
self._reactor_thread.daemon = True
self._reactor_thread.start()
def send_message(self, body):
if not self._check_connection():
return
log.msg("Queing send")
self._send_queue.put(body)
reactor.callFromThread(self._dispatch)
def _check_connection(self):
if not self._connected_event.wait(timeout=10):
log.err("Unable to connect to server")
self.close()
return False
return True
def _dispatch(self):
log.msg("Dispatching")
while True:
try:
body = self._send_queue.get(block=False)
except Queue.Empty:
break
self.factory.protocol_instance.sendMessage(body)
def close(self):
reactor.callFromThread(reactor.stop)
import time
def Ppippo(coda):
while True:
coda.send_message('YOOOOOOOO')
time.sleep(5)
if __name__ == '__main__':
ws_setting = {'host':'', 'port':}
client = BaseWBClient(ws_setting)
t1 = threading.Thread(client.connect())
t11 = threading.Thread(Ppippo(client))
t11.start()
t1.start()
Предыдущий код работает нормально, но мне нужно перевести его для работы с протоколом WAMP, вставленным в веб-сокет.
Кто-нибудь знает, как я решаю?
thread1.join()
вниз, чтобы быть сthread2.join()
. В своем текущем местоположении он скажет основному потоку подождать, пока поток 1 не умрет. Поскольку у вас нет возможности убить поток (без уничтожения всего процесса с помощью Ctrl-C), второй поток никогда не будет создан. - person Aaron D   schedule 09.02.2015.run()
. Выjoin()
создаете поток в конце своей основной функции, чтобы дать потокам время завершить свое выполнение до закрытия основного приложения. Таким образом, вам нужно иметь возможность для потоков завершить свою задачу. - person Aaron D   schedule 09.02.2015