04消息队列zmq的发布者-订阅者的计算π的简易程序。

# 小常识：测算π的在其中一个方式是	，任意的向一个周长为n的方形中撒黄豆。
# 随后看这种黄豆是不是在以n为半经的四分之一圆内
，正方形面积：n*n，四分之一圆的面积：π*n*n/4
# 因而落在四分之一圆内的几率为π/4	，那样大家就能算出π的值	。
# 大家这一程序流程是来叙述zmq公布-订阅全过程的步骤，

# 根据测算π数值事例来开展测算
。
# 最先发布者bitsource随机生成一个字符串数组
，字符串数组双数位为纵坐标，合数位为横坐标轴。
# 一个字符串数组表明一个点	，将合数位变换为横坐标轴
，将双数位变换为纵坐标，随后根据测算
# 这一点至起点的间距	，来分辨是不是为以半经为B的圆内
，根据很多的仿真模拟，来开展测算π	。



import random, threading, time, zmq
B = 32  # number of bits of precision in each random integer

def ones_and_zeros(digits):
    """Express `n` in at least `d` binary digits, with no special prefix."""
    # getrandbits()
    # 方式回到特定尺寸（以位为企业）的整数金额。
    return bin(random.getrandbits(digits)).lstrip('0b').zfill(digits)

# 发布者
def bitsource(zcontext, url):
    # 公布订阅涵数	，发布者
。
    """Produce random points in the unit square."""
    # 造就一个对象。发布者对象
    zsock = zcontext.socket(zmq.PUB)
    # 关联URL
    zsock.bind(url)
    while True:
        # 不断持续的一直推送这儿是32位系统的二进制标识符	。
        # 这儿一个字符串数组意味着一个点
。
        zsock.send_string(ones_and_zeros(B * 2))
        # 间断0.01秒。
        time.sleep(0.01)
# 订阅者1
def always_yes(zcontext, in_url, out_url):
    """Coordinates in the lower-left quadrant are inside the unit circle."""
    # 造就一个订阅者对象	。
    isock = zcontext.socket(zmq.SUB)
    # 联接发布者的URL
    isock.connect(in_url)
    # 设定过虑标准
，接收00开始的
    # 假如开始为00得话，那麼这一点x	、y值都不容易超出半经的一半
    # 一定在四分之一圆内
。
    isock.setsockopt(zmq.SUBSCRIBE, b'00')
    # 消息推送。
    osock = zcontext.socket(zmq.PUSH)
    # 联接接收消息推送的URL
    osock.connect(out_url)
    while True:
        # 接收订阅者的信息	。
        isock.recv_string()
        # 消息推送给发布者信息
。因而大家这儿立即推送Y
        osock.send_string('Y')
# 订阅者2
def judge(zcontext, in_url, pythagoras_url, out_url):
    """Determine whether each input coordinate is inside the unit circle."""
    #     # 造就一个订阅者对象。
    isock = zcontext.socket(zmq.SUB)
    # 联接URL
    isock.connect(in_url)
    # 设定接收订阅的过虑标准	。
    for prefix in b'01', b'10', b'11':
        isock.setsockopt(zmq.SUBSCRIBE, prefix)
    # 设定一个回应对象
。
    psock = zcontext.socket(zmq.REQ)
    psock.connect(pythagoras_url)
    # 设定一个消息推送对象。
    osock = zcontext.socket(zmq.PUSH)
    osock.connect(out_url)
    # 这儿用了勾股定理	，是2个落在纵坐标上的点
，平方和	。
    unit = 2 ** (B * 2)
    # 这儿必须测算是不是在四分之一圆内。
    while True:
        # 接收发布者的信息
。
        bits = isock.recv_string()
        # 获取这一点的x座标，y座标。
        n, m = int(bits[::2], 2), int(bits[1::2], 2)
        # 发给顾客要求端
        psock.send_json((n, m))
        # 随后接纳顾客要求端推送回来解决过的数据信息	。
        sumsquares = psock.recv_json()
        # 分辨是不是在圆内
。
        osock.send_string('Y' if sumsquares < unit else 'N')
# 要求端	，
def pythagoras(zcontext, url):
    """Return the sum-of-squares of number sequences."""
    zsock = zcontext.socket(zmq.REP)
    zsock.bind(url)
    while True:
        # 这儿先要求数据信息，随后将要求的数据信息开展解决推送出来。
        numbers = zsock.recv_json()
        zsock.send_json(sum(n * n for n in numbers))
# 归纳
，开展测算π的值	。
def tally(zcontext, url):
    """Tally how many points fall within the unit circle, and print pi."""
    zsock = zcontext.socket(zmq.PULL)
    zsock.bind(url)
    # 这儿是假如接纳到一个Y p 4，接纳到一个N 	，q   1
    # 随后测算比率
，这一便是大家算出去的π的值
。
    p = q = 0
    while True:
        decision = zsock.recv_string()
        q  = 1
        if decision == 'Y':
            p  = 4
        print(decision, p / q)

# 大家应用线程同步的方法，上面的每一个涵数开一个进程	，。
def start_thread(function, *args):
    thread = threading.Thread(target=function, args=args)
    thread.daemon = True  # so you can easily Ctrl-C the whole program
    thread.start()

def main(zcontext):
    pubsub = 'tcp://127.0.0.1:6700'
    reqrep = 'tcp://127.0.0.1:6701'
    pushpull = 'tcp://127.0.0.1:6702'
    start_thread(bitsource, zcontext, pubsub)
    start_thread(always_yes, zcontext, pubsub, pushpull)
    start_thread(judge, zcontext, pubsub, reqrep, pushpull)
    start_thread(pythagoras, zcontext, reqrep)
    start_thread(tally, zcontext, pushpull)
    # 这个是主线任务程
，主线任务程完毕，别的的线程同步也就需要完毕	。
    time.sleep(30)

if __name__ == '__main__':
    main(zmq.Context())