<!DOCTYPE html PUBLIC "-//W3C//DTD XHTML 1.0 Transitional//EN" "http://www.w3.org/TR/xhtml1/DTD/xhtml1-strict.dtd"> <html><head><meta http-equiv="Cont

简介: #-*- coding:utf-8 -*-import Queueimport threadingimport timeimport jsonimport sysimport signalimport randomreload( sys )sys.
#-*- coding:utf-8 -*-

import Queue
import threading
import time
import json
import sys
import signal
import random
reload( sys )
sys.setdefaultencoding('utf-8')

class Enum(set):
    def __getattr__(self, name):
        if name in self:
            return name
        else:
            raise AttributeError
State = Enum(['NORMAL', 'UPDATE', 'STOP'])

engine_do = True
def handler(signum, frame):
    print 'receive signal: %s' % signum
    global engine_do
    engine_do = False

signal.signal(signal.SIGINT, handler)
signal.signal(signal.SIGTERM, handler)

class Consumer(threading.Thread):
    def __init__(self, queue):
        threading.Thread.__init__(self)
        self.queue = queue
        self.do = True

    def stop(self):
        self.do = False
        print 'change consumer.do to False'
    
    def run(self):
        print 'Create new consumer thread, id: %s' % self.ident
        while self.do:
            messages = []
            result = []
            msg = random.randint(0,100)
            self.queue.put(msg)
        print 'Consumer thread will exit.'
        
class Producer(threading.Thread):
    def __init__(self, queue):
        threading.Thread.__init__(self)
        self.queue = queue
        self.msgs = Queue.Queue()
        self.state = State.NORMAL
        self.do = True

    def stop(self):
        self.do = False
        self.state = State.STOP

    def run(self):
        while self.do:
            if self.state == State.NORMAL:
                if not self.queue.empty():
                    data = self.queue.get()
                    print 'Producer get data: %s' % data
                else:
                    print 'data queue is empty, sleep 5 seconds.'
                    time.sleep(5)
            elif self.state == State.STOP:
                while not self.queue.empty():
                    data = self.queue.get()
                    print 'Producer get data: %s' % data
        print 'Producer thread will exit.'

class Engine():
    def __init__(self):
        # 在获取所有的topic并初始化连接
        # 初始化消费Queue中数据的线程
        self.queue = Queue.Queue()
        self.threads_consumer = []
        self.threads_producer = []


    def run(self):
        # 启动Consumer线程
        for i in xrange(10):
            consumer = Consumer(self.queue)
            consumer.start()
            self.threads_consumer.append(consumer)
        producer = Producer(self.queue)
        self.threads_producer.append(producer)
        producer.start()
        while True:
            time.sleep(5)
            print engine_do
            if not engine_do:
                print 'engine will exit...'
                print 'first stop consumer threads'
                for consumer in self.threads_consumer:
                    consumer.stop()
                for consumer in self.threads_consumer:
                    consumer.join()
                print 'all consumer threads are done.'
                print 'second stop producer threads...'
                for producer in self.threads_producer:
                    producer.stop()
                for producer in self.threads_producer:
                    producer.join()
                print 'all producer threads are done.'
                break
        print 'All threads are not alive, main thread will exit.'
        return 

if __name__=='__main__':
    engine = Engine()
    engine.run()

目录
相关文章
|
Web App开发 前端开发
<!DOCTYPE html PUBLIC "-//W3C//DTD XHTML 1.0 Transitional//EN" "http://www.w3.org/TR/xhtml1/DTD/xhtml1-strict.dtd"> <html><head><meta http-equiv="Cont
zookeeper watch的定义如下:watch事件是一次性触发器,当watch监视的数据发生变化时,通知设置了该watch的client,即watcher。
899 0
|
3天前
|
NoSQL Cloud Native Redis
Redis核心开发者的新征程:阿里云与Valkey社区的技术融合与创新
阿里云瑶池数据库团队后续将持续参与Valkey社区,如过往在Redis社区一样耕耘,为开源社区作出持续贡献。
Redis核心开发者的新征程:阿里云与Valkey社区的技术融合与创新
|
4天前
|
弹性计算 安全 API
访问控制(RAM)|云上安全使用AccessKey的最佳实践
集中管控AK/SK的生命周期,可以极大降低AK/SK管理和使用成本,同时通过加密和轮转的方式,保证AK/SK的安全使用,本次分享为您介绍产品原理,以及具体的使用步骤。
101786 0
|
5天前
|
SQL 关系型数据库 分布式数据库
Doodle Jump — 使用Flutter&Flame开发游戏真不错!
用Flutter&Flame开发游戏是一种什么体验?最近网上冲浪的时候,我偶然发现了一个国外的游戏网站,类似于国内的4399。在浏览时,我遇到了一款经典的小游戏:Doodle Jump...
|
8天前
|
SQL 存储 JSON
Flink+Paimon+Hologres 构建实时湖仓数据分析
本文整理自阿里云高级专家喻良,在 Flink Forward Asia 2023 主会场的分享。
71265 0
Flink+Paimon+Hologres 构建实时湖仓数据分析
|
12天前
|
弹性计算 运维 安全
访问控制(RAM)|云上程序使用临时凭证的最佳实践
STS临时访问凭证是阿里云提供的一种临时访问权限管理服务,通过STS获取可以自定义时效和访问权限的临时身份凭证,减少长期访问密钥(AccessKey)泄露的风险。本文将为您介绍产品原理,以及具体的使用步骤。
151035 4
|
10天前
|
数据采集 存储 运维
提升团队工程交付能力,从“看见”工程活动和研发模式开始
本文从统一工程交付的概念模型开始,介绍了如何将应用交付的模式显式地定义出来,并通过工具平台落地。
119991 57
|
11天前
|
监控 负载均衡 Java
深入探究Java微服务架构:Spring Cloud概论
**摘要:** 本文深入探讨了Java微服务架构中的Spring Cloud,解释了微服务架构如何解决传统单体架构的局限性,如松耦合、独立部署、可伸缩性和容错性。Spring Cloud作为一个基于Spring Boot的开源框架,提供了服务注册与发现、负载均衡、断路器、配置中心、API网关等组件,简化了微服务的开发、部署和管理。文章详细介绍了Spring Cloud的核心模块,如Eureka、Ribbon、Hystrix、Config、Zuul和Sleuth,并通过一个电商微服务系统的实战案例展示了如何使用Spring Cloud构建微服务应用。
103505 8
|
12天前
|
人工智能 Serverless 对象存储
让你的文档从静态展示到一键部署可操作验证
通过函数计算的能力让阿里云的文档从静态展示升级为动态可操作验证,用户在文档中单击一键部署可快速完成代码的部署及测试。这一改变已在函数计算的活动沙龙中得到用户的认可。
120874 229