[Python] python zeromq Parallel Pipeline multiple consumers (workers)

Hello together!

I would like to combine data from different python programs via zeromq. I think for that job the best solution would be parallel pipelines as are described in this example at the ponit "3. Parallel Pipeline (parallel pipeline mode)".

So in the beginning I wanted to test the functionality with just very simple examples. That for I've just used the three patterns producer, consumer and resultcollector that you can find on the example. I just made small changes:

Producer

import zmq

context = zmq.Context()
socket = context.socket(zmq.PUSH)
socket.bind("tcp://*:5566")

work_message = "start working"
socket.send_json(work_message)


Consumer_1

import random
import zmq


context = zmq.Context()
consumer_id = 199

#receiving work
consumer_receiver = context.socket(zmq.PULL)
consumer_receiver.connect("tcp://localhost:5566")

#forewarding results
consumer_sender = context.socket(zmq.PUSH)
consumer_sender.bind("tcp://*:5500")

msg = consumer_receiver.recv_json()

for i in (0, 100):
if msg == "start working":
data = "id: " + str(consumer_id) + "; Hello"
consumer_sender.send_json(data)


Result Collector

#coding:utf-8

import zmq

context = zmq.Context()
result_receiver = context.socket(zmq.PULL)
result_receiver.connect("tcp://localhost:5599")

result = result_receiver.recv_json()

collected_data = []

for i in (0, 10000):
collected_data.append(result)

print(collected_data)


So the main communication between them is working. But now I've tried to add another consumer (worker) who delivers data. It should be possible, as you can see in the example I've linked. Here you can see my second consumer (worker):

Consumer_2

import random
import zmq


context = zmq.Context()
consumer_id = 10

#receiving work
consumer_receiver = context.socket(zmq.PULL)
consumer_receiver.connect("tcp://localhost:5566")

#forewarding results
consumer_sender = context.socket(zmq.PUSH)
consumer_sender.bind("tcp://*:5599")

msg = consumer_receiver.recv_json()

for i in (0, 100):
if msg == "start working":
data = "id: " + str(consumer_id) + "; World"
consumer_sender.send_json(data)


It's pretty much the same as consumer_1. But when I want to run both consumers i get the error message "ZMQError: Adress in use":

---------------------------------------------------------------------------
ZMQError Traceback (most recent call last)
<ipython-input-1-af8297fe1137> in <module>
12 #forewarding results
13 consumer_sender = context.socket(zmq.PUSH)
---> 14 consumer_sender.bind("tcp://*:5599")
15
16 msg = consumer_receiver.recv_json()

~\Anaconda3\lib\site-packages\zmq\sugar\socket.py in bind(self, addr)
171
172 """
--> 173 super().bind(addr)
174 return self._bind_cm(addr)
175

zmq/backend/cython/socket.pyx in zmq.backend.cython.socket.Socket.bind()

~\Anaconda3\lib\site-packages\zmq\backend\cython\checkrc.pxd in zmq.backend.cython.checkrc._check_rc()

ZMQError: Address in use


Where is my fault? It has to be possible to push with different programms to a single port (as in the example port 5599) or am I wrong? I am pretty new to programming at all so sorry if you think that this is a stupid question.

It would be very nice if you guys could help me out here.

Thanks a lot!

Continue reading...