经验首页 前端设计 程序设计 Java相关 移动开发 数据库/运维 软件/图像 大数据/云计算 其他经验
当前位置：技术经验 » 大数据/云/AI » Apache Kafka » 查看文章
大数据之路【第十篇】：kafka消息系统
来源：cnblogs　　作者：Simon92　　时间：2019/8/30 9:13:32　　对本文有异议

一、简介

1、简介

简 介
• Kafka是Linkedin于2010年12月份开源的消息系统
• 一种分布式的、基于发布/订阅的消息系统

2、特点

– 消息持久化：通过O(1)的磁盘数据结构提供数据的持久化
– 高吞吐量：每秒百万级的消息读写
– 分布式：扩展能力强
– 多客户端支持：java、php、python、c++ ……
– 实时性：生产者生产的message立即被消费者可见

3、基本组件

• Broker：每一台机器叫一个Broker
• Producer：日志消息生产者，用来写数据
• Consumer：消息的消费者，用来读数据
• Topic：不同消费者去指定的Topic中读，不同的生产者往不同的Topic中写
• Partition：新版本才支持Partition，在Topic基础上做了进一步区分分层

好处二：动态导入模块（基于反射当前模块成员）

注意：

• Kafka内部是分布式的、一个Kafka集群通常包括多个Broker
• 负载均衡：将Topic分成多个分区，每个Broker存储一个或多个Partition
• 多个Producer和Consumer同时生产和消费消息

4、Topic

• 一个Topic是一个用于发布消息的分类或feed名，kafka集群使用分区的日志，每个分区都是有顺序且不变的消息序列。

• commit的log可以不断追加。消息在每个分区中都分配了一个叫offset的id序列来唯一识别分区中的消息

 

• 举例：若创建topic1和topic2两个topic，且分别有13个和19个分区，则整个集群上会相应会生成共32个文件夹

注意：

• 无论发布的消息是否被消费，kafka都会持久化一定时间（可配置）。

• 在每个消费者都持久化这个offset在日志中。通常消费者读消息时会使offset值线性的增长，但实际上其位置是由消费者控制，它可以按任意顺序来消费消息。

比如复位到老的offset来重新处理。
• 每个分区代表一个并行单元。

 

5、Message

  1. • message（消息）是通信的基本单位，每个producer可以向一个topic（主题）
  2. 发布一些消息。如果consumer订阅了这个主题，那么新发布的消息就会广播给
  3. 这些consumer。
  4. • message format：
  5. – message length : 4 bytes (value: 1+4+n)
  6. – "magic" value : 1 byte
  7. – crc : 4 bytes
  8. – payload : n bytes

6、Producer

• 生产者可以发布数据到它指定的topic中，并可以指定在topic里哪些消息分配到哪些分区（比如简单的轮流分发各个分区或通过指定分区语义分配key到对应分

区）
• 生产者直接把消息发送给对应分区的broker，而不需要任何路由层。
• 批处理发送，当message积累到一定数量或等待一定时间后进行发送。

7、Consumer

• 一种更抽象的消费方式：消费组（consumer group）
• 该方式包含了传统的queue和发布订阅方式
    –  首先消费者标记自己一个消费组名。消息将投递到每个消费组中的某一个消费者实例上。
    –  如果所有的消费者实例都有相同的消费组，这样就像传统的queue方式。
    –  如果所有的消费者实例都有不同的消费组，这样就像传统的发布订阅方式。
    –  消费组就好比是个逻辑的订阅者，每个订阅者由许多消费者实例构成(用于扩展或容错)。
• 相对于传统的消息系统，kafka拥有更强壮的顺序保证。
• 由于topic采用了分区，可在多Consumer进程操作时保证顺序性和负载均衡。

 

三、包装标准类型

包装：python为大家提供了标准数据类型，以及丰富的内置方法，其实在很多场景下我们都需要基于标准数据类型来定制我们自己的数据类型，新增/改写方法，这就用到了我们刚学的继承/派生知识 其他的标准类型均可以通过下面的方式进行二次加工

  1. class List(list):
  2.     def append(self,p_object):
  3.         if type(p_object) is str:
  4.             # self.append(p_object)
  5.             super().append(p_object)
  6.         else:
  7.             print('只能添加字符串类型')
  8.     def show_middle(self):
  9.         mid_index=int(len(self)/2)
  10.         return self[mid_index]
  13. # l2=List('hello world')
  14. # print(l2,type(l2))
  16. l1=List('helloworld')
  17. # print(l1,type(l1))
  18. l1.append(123456)
  19. l1.append('sb')
  20. print(l1)

 

四、isinstance(obj,cls)和issubclass(sub,super)

isinstance(obj,cls)检查是否obj是否是cls的对象

  1. class Foo(object):
  2.       pass
  4. obj  = Foo()
  6. isinstance(obj,  Foo)

issubclass(sub, super)检查sub类是否是 super 类的派生类

  1. class Foo:
  2.     pass
  5. class Bar(Foo):
  6.     pass
  7. f1=Foo()
  8. print(isinstance(f1,Foo))
  9. print(issubclass(Bar,Foo))

五、__getattribute__

  1. class Foo:
  2.     def __init__(self,x):
  3.         self.x=x
  5.     def __getattr__(self, item):
  6.         print('执行的是getattr')
  7.     def __getattribute__(self, item):
  8.         print('执行的是getattribute')
  9.         raise AttributeError('抛出异常了')
  11. f1=Foo(11)
  12. # f1.x
  13. f1.xxxxxx  #不存在的属性访问，触发__getattr__

六、__setitem__,__getitem__,__delitem__

  1. class Foo:
  2.     def __getitem__(self, item):
  3.         print('getitem')
  4.         # retun self.__dict__
  6.     def __setitem__(self, key, value):
  7.         print('setitem')
  8.         self.__dict__[key]=value
  10.     def __delitem__(self, key):
  11.         print('delitem')
  12.         self.__dict__.pop(key)
  15. f1=Foo()
  16. print(f1.__dict__)
  17. # f1.name='simon'
  18. f1['name']='simon'
  19. f1['age']=28
  21. print('===========>',f1.__dict__)
  23. # del f1.name
  24. # print(f1.__dict__)
  25. # print(f1.age)
  26. del f1['name']
  27. print(f1.__dict__)

七、__str__,__repr__

  1. #####str#######
  2. class Foo:
  3.     def __init__(self,name,age):
  4.         self.name=name
  5.         self.age=age
  7.     def __str__(self):
  8.         return '名字是【%s】 年龄是【%s】' %(self.name,self.age)
  9. f1=Foo('simon',18)
  10. print(f1) #--str(f1)-->f1.__str__()
  11. x=str(f1)
  12. print(x)
  14. ######repr######## ,当str与repr共存是，优先使用str
  15. class Foo:
  16.     def __init__(self,name,age):
  17.         self.name=name
  18.         self.age=age
  19.     # def __str__(self):
  20.     #     return '这是str'
  22.     def __repr__(self):
  23.         return '名字是【%s】 年龄是【%s】' %(self.name,self.age)
  24. f1=Foo('simon',20)
  25. #repr(f1) ----->f1.__repr__()
  26. print(f1) #str(f1) ---->> f1.__str__() ------>f1.__repr__()

备注：

  1. '''
  2. str函数或者print函数--->obj.__str__()
  3. repr或者交互式解释器--->obj.__repr__()
  4. 如果__str__没有被定义,那么就会使用__repr__来代替输出
  5. 注意:这俩方法的返回值必须是字符串,否则抛出异常
  6. '''

八、__format__

  1. format_dic={
  2.     'ymd':'{0.year}{0.mon}{0.day}',
  3.     'm-d-y':'{0.mon}-{0.day}-{0.year}',
  4.     'y:m:d':'{0.year}:{0.mon}:{0.day}'
  6. }
  7. class Date:
  8.     def __init__(self,year,mon,day):
  9.         self.year=year
  10.         self.mon=mon
  11.         self.day=day
  12.     def __format__(self, format_spec):
  13.         print('我要执行啦')
  14.         print('------->',format_spec)
  15.         if not format_spec or format_spec not in format_dic:
  16.             format_spec='ymd'
  17.         fm = format_dic[format_spec]
  18.         return fm.format(self)
  19. d1=Date(2018,12,30)
  20. format(d1) #d1.__format__()
  21. print(format(d1))
  22. print(format(d1,'ymd'))
  23. print(format(d1,'y:m:d'))
  24. print(format(d1,'m-d-y'))
  25. print(format(d1,'fsdrerewr'))

九、__slots__

  1. 1.__slots__是什么:是一个类变量,变量值可以是列表,元祖,或者可迭代对象,也可以是一个字符串(意味着所有实例只有一个数据属性)
  2. 2.引子:使用点来访问属性本质就是在访问类或者对象的__dict__属性字典(类的字典是共享的,而每个实例的是独立的)
  3. 3.为何使用__slots__:字典会占用大量内存,如果你有一个属性很少的类,但是有很多实例,为了节省内存可以使用__slots__取代实例的__dict__
  4. 当你定义__slots__后,__slots__就会为实例使用一种更加紧凑的内部表示。实例通过一个很小的固定大小的数组来构建,而不是为每个实例定义一个
  5. 字典,这跟元组或列表很类似。在__slots__中列出的属性名在内部被映射到这个数组的指定小标上。使用__slots__一个不好的地方就是我们不能再给
  6. 实例添加新的属性了,只能使用在__slots__中定义的那些属性名。
  7. 4.注意事项:__slots__的很多特性都依赖于普通的基于字典的实现。另外,定义了__slots__后的类不再 支持一些普通类特性了,比如多继承。大多数情况下,你应该
  8. 只在那些经常被使用到 的用作数据结构的类上定义__slots__比如在程序中需要创建某个类的几百万个实例对象 。
  9. 关于__slots__的一个常见误区是它可以作为一个封装工具来防止用户给实例增加新的属性。尽管使用__slots__可以达到这样的目的,但是这个并不是它的初衷。           更多的是用来作为一个内存优化工具。
  1. class Foo:
  2.     __slots__=['name','age']
  4. f1=Foo()
  5. f1.name='alex'
  6. f1.age=18
  7. print(f1.__slots__)
  9. f2=Foo()
  10. f2.name='egon'
  11. f2.age=19
  12. print(f2.__slots__)
  14. print(Foo.__dict__)
  15. #f1与f2都没有属性字典__dict__了,统一归__slots__管,节省内存

十、__doc__

  1. class Foo:
  2.     '我是描述信息'
  3.     pass
  5. class Bar(Foo):
  6.     pass
  7. # print(Foo.__doc__) #该属性无法继承给子类
  9. print(Bar.__dict__)
  10. print(Bar.__doc__) #该属性无法继承给子类

十一、__module__和__class__

__module__表示当前操作的对象在哪个模块

—class__ 表示当前操作的对象的类是什么

lib/simon.py

  1. #！/usr/bin/env python
  2. # _*_ coding:utf-8 _*_
  3.  
  4. class C:
  5.     def __init__(self):
  6.         self.name = 'simon'
  7.

aa.py

  1. from lib.simon import C
  3. obj=C()
  4. print(obj.__module__) #输出 lib.simon,即：输出模块
  5. print(obj.__class__)  #输出lib.simon.C,即：输出类

十二、__del__

  1. class Foo:
  2.     def __init__(self,name):
  3.         self.name=name
  4.     def __del__(self):
  5.         print('我执行啦')
  7. f1=Foo('simon')
  9. # del f1    #删除实例会触发__del__
  10. del f1.name #删除实例的属性不会触发__del__
  11. print('--------------------->')
  13. #程序运行完毕会自动回收内存，触发__del__

十三、__call__

对象后面加括号，触发执行。

注：构造方法的执行是由创建对象触发的，即：对象 = 类名() ；而对于 __call__ 方法的执行是由对象后加括号触发的，即：对象() 或者 类()()

  1. class Foo:
  2.     def __call__(self, *args, **kwargs):
  3.         print('实例执行啦　ｏｂｊ（）')
  5. f1=Foo()
  7. f1() #f1的类Foo 下的__call__
  9. Foo() #Foo的类 xxx下的__call__

十四、__next__和__iter__实现迭代器协议

1、迭代器协议是指：对象必须提供一个Next方法，执行该方法要么返回迭代中的下一项，要么就引起一个StopIteration异常，以终止迭代(只能往后走不能往前退)

2、可迭代对象:实现了迭代器协议的对象（如何实现：对象内部定义一个__iter__()方法）

3、协议是一种约定，可迭代对象实现了迭代器协议，python内部工具（如for循环，sum,min,max函数等）使用迭代器协议访问对象。

  1. class Foo:
  2.     def __init__(self,n):
  3.         self.n=n
  4.     def __iter__(self):
  5.         return self
  6.     def __next__(self):
  7.         if self.n == 13:
  8.             raise StopIteration('终止了')
  9.         self.n+=1
  10.         return self.n
  11. # l=list('simon')
  12. # for i in l:
  13. #     print(i)
  15. f1=Foo(10)
  16. # print(f1.__next__())
  17. # print(f1.__next__())
  18. # print(f1.__next__())
  19. for i in f1: #iter(f1)------------>f1.__iter__()
  20.     print(i)

斐波那契数列：

  1. class Fib:
  2.     def __init__(self):
  3.         self._a=1
  4.         self._b=1
  6.     def __iter__(self):
  7.         return self
  8.     def __next__(self):
  9.         if self._a > 100:
  10.             raise StopIteration('终止了')
  11.         self._a,self._b=self._b,self._a + self._b
  12.         return self._a
  14. f1=Fib()
  15. print(next(f1))
  16. print(next(f1))
  17. print(next(f1))
  18. print(next(f1))
  19. print(next(f1))
  20. print('============================')
  22. for i in f1:
  23.     print(i)

十五、描述符(_get_,_set_,_delete_)

1、什么是描述符

描述符本质就是一个新式类，在这个新式类中，至少实现了__get__(),__set__(),__delete__()中的一个，这也被称为描述符协议

__get__():调用一个属性时，触发

__set__():为一个属性赋值时，触发

__delete__():采用del删除属性时，触发

  1. class Foo:
  2.     def __get__(self, instance, owner):
  3.         print('=====>get方法')
  5.     def __set__(self, instance, value):
  6.         print('=====>set方法')
  8.     def __delete__(self, instance):
  9.         print('=====>delete方法')
  11. class Bar:
  12.     x=Foo()  #在何地？
  14. #在何时
  15. b1=Bar()
  16. b1.x
  18. b1.x=1
  20. del b1.x
  21. f1=Foo()
  22. f1.name='simon'
  23. print(f1.name)
  1.  
  1. 类属性>数据描述符
  1. #描述符Str
  2. class Str:
  3.     def __get__(self, instance, owner):
  4.         print('Str调用')
  5.     def __set__(self, instance, value):
  6.         print('Str设置...')
  7.     def __delete__(self, instance):
  8.         print('Str删除...')
  10. class People:
  11.     name=Str()
  12.     def __init__(self,name,age): #name被Str类代理,age被Int类代理,
  13.         self.name=name
  14.         self.age=age
  17. #基于上面的演示,我们已经知道,在一个类中定义描述符它就是一个类属性,存在于类的属性字典中,而不是实例的属性字典
  19. #那既然描述符被定义成了一个类属性,直接通过类名也一定可以调用吧,没错
  20. People.name #恩,调用类属性name,本质就是在调用描述符Str,触发了__get__()
  22. People.name='egon' #那赋值呢,我去,并没有触发__set__()
  23. del People.name #赶紧试试del,我去,也没有触发__delete__()
  24. #结论:描述符对类没有作用-------->傻逼到家的结论
  26. '''
  27. 原因:描述符在使用时被定义成另外一个类的类属性,因而类属性比二次加工的描述符伪装而来的类属性有更高的优先级
  28. People.name #恩,调用类属性name,找不到就去找描述符伪装的类属性name,触发了__get__()
  30. People.name='egon' #那赋值呢,直接赋值了一个类属性,它拥有更高的优先级,相当于覆盖了描述符,肯定不会触发描述符的__set__()
  31. del People.name #同上
  32. '''

数据描述符>实例属性

  1. #描述符Str
  2. class Str:
  3.     def __get__(self, instance, owner):
  4.         print('Str调用')
  5.     def __set__(self, instance, value):
  6.         print('Str设置...')
  7.     def __delete__(self, instance):
  8.         print('Str删除...')
  10. class People:
  11.     name=Str()
  12.     def __init__(self,name,age): #name被Str类代理,age被Int类代理,
  13.         self.name=name
  14.         self.age=age
  17. p1=People('egon',18)
  19. #如果描述符是一个数据描述符(即有__get__又有__set__),那么p1.name的调用与赋值都是触发描述符的操作,于p1本身无关了,相当于覆盖了实例的属性
  20. p1.name='egonnnnnn'
  21. p1.name
  22. print(p1.__dict__)#实例的属性字典中没有name,因为name是一个数据描述符,优先级高于实例属性,查看/赋值/删除都是跟描述符有关,与实例无关了
  23. del p1.name

实例属性>非数据描述符

  1. class Foo:
  2.     def func(self):
  3.         print('我胡汉三又回来了')
  4. f1=Foo()
  5. f1.func() #调用类的方法,也可以说是调用非数据描述符
  6. #函数是一个非数据描述符对象(一切皆对象么)
  7. print(dir(Foo.func))
  8. print(hasattr(Foo.func,'__set__'))
  9. print(hasattr(Foo.func,'__get__'))
  10. print(hasattr(Foo.func,'__delete__'))
  11. #有人可能会问,描述符不都是类么,函数怎么算也应该是一个对象啊,怎么就是描述符了
  12. #笨蛋哥,描述符是类没问题,描述符在应用的时候不都是实例化成一个类属性么
  13. #函数就是一个由非描述符类实例化得到的对象
  14. #没错，字符串也一样
  17. f1.func='这是实例属性啊'
  18. print(f1.func)
  20. del f1.func #删掉了非数据
  21. f1.func()

再次验证：实例属性>非数据描述符

  1. class Foo:
  2.     def __set__(self, instance, value):
  3.         print('set')
  4.     def __get__(self, instance, owner):
  5.         print('get')
  6. class Room:
  7.     name=Foo()
  8.     def __init__(self,name,width,length):
  9.         self.name=name
  10.         self.width=width
  11.         self.length=length
  14. #name是一个数据描述符,因为name=Foo()而Foo实现了get和set方法,因而比实例属性有更高的优先级
  15. #对实例的属性操作,触发的都是描述符的
  16. r1=Room('厕所',1,1)
  17. r1.name
  18. r1.name='厨房'
  22. class Foo:
  23.     def __get__(self, instance, owner):
  24.         print('get')
  25. class Room:
  26.     name=Foo()
  27.     def __init__(self,name,width,length):
  28.         self.name=name
  29.         self.width=width
  30.         self.length=length
  33. #name是一个非数据描述符,因为name=Foo()而Foo没有实现set方法,因而比实例属性有更低的优先级
  34. #对实例的属性操作,触发的都是实例自己的
  35. r1=Room('厕所',1,1)
  36. r1.name
  37. r1.name='厨房'

 总结：

1、描述符本身应该定义成新式类，被代理的类也应该是新式类

2、必须把描述符定义成这个类的代理，不能为定义到构造函数中

3、要严格遵循优先级，优先级由高到低分别

 

十六、__enter__和__exit__

操作文件对象的时候：

  1. with open('text.txt','r') as f:
  2.         '代码块'

上述叫做上下文管理协议，即with语句，为了让一个对象兼容with语句，必须在这个对象的类中声明__enter__和__exit__方法

  1. class Foo:
  2.     def __init__(self,name):
  3.         self.name=name
  5.     def __enter__(self):
  6.         print('执行enter')
  7.         return self
  9.     def __exit__(self, exc_type, exc_val, exc_tb):
  10.         print('执行exit')
  12. # f=Open('a.txt')
  13. with Foo('a.txt') as f:
  14.     print(f)
  15.     print(f.name)
  16.     print('00000000000000000000')

上述代码分析：

  1. class Foo:
  2.     def __init__(self,name):
  3.         self.name=name
  5.     def __enter__(self):
  6.         print('执行enter')
  7.         return self
  9.     def __exit__(self, exc_type, exc_val, exc_tb):
  10.         print('执行exit')
  11.         print(exc_type)
  12.         print(exc_val)
  13.         print(exc_tb)
  14.         return True
  16. # f=Open('a.txt')
  17. with Foo('a.txt') as f:
  18.     print(f)
  19.     print(asfdreevergewafa)  #触发__exit__
  20.     print(f.name)
  21. print('00000000000000000000')
  23. #with obj as f:
  24.     '代码块'
  26. 1、with obj ---->>触发obj.__enter__(),拿到返回值
  27. 2、as f--------->f=返回值、
  28. 3、with obj as f 等同于  f=obj.__enter__()
  29. 4、执行代码块
  30. 一：没有异常的情况下，整个代码块运行完毕后去触发__exit__,它的三个参数都为None
  31. 二：有异常的情况下，从异常出现的位置直接触发__exit__
  32.     a: 如果__exit__的返回值为True，代表吞掉了异常
  33.     b: 如果__exit__的返回值不为True，代表了吐出了异常
  34.     c: __exit__的运行完毕就代表了整个with语句的执行完毕

总结：

1.使用with语句的目的就是把代码块放入with中执行，with结束后，自动完成清理工作，无须手动干预

2.在需要管理一些资源比如文件，网络连接和锁的编程环境中，可以在__exit__中定制自动释放资源的机制，你无须再去关系这个问题，这将大有用处

 

原文链接：http://www.cnblogs.com/hackerer/p/11430484.html

 友情链接：直通硅谷 　点职佳 　北美留学生论坛

Apache Kafka热门文章

  • Kafka监控工具汇总
  • Kraft模式下Kafka脚本的使用
  • Kafka web console安装
  • kafka_2.11-2.0.0_安装部署
  • ELK日志方案--使用Filebeat收集日志并输出到Kafka
  • Kafka相关内容总结（存储和性能）
  • 一次flume exec source采集日志到kafka因为单条日志数据非常大同步失败的踩坑带来的思考
  • Kafka使用jmxtrans+influxdb+grafana监控JMX指标
  • Micronaut 微服务中使用 Kafka
  • 什么是Kafka？

    • Apache Kafka推荐文章

  • 异源数据同步 → DataX 为什么要支持 kafka？
  • 解锁强强组合: 使用 Kafka + ClickHouse 快速搭建流数据实时处理平台(DoubleCloud 博客)
  • Kafka 线上性能调优
  • Kafka原理剖析之「位点提交」
  • Kafka 社区KIP-382中文译文（MirrorMaker2/集群复制/高可用/灾难恢复）
  • kafka源码阅读之MacBook Pro M1搭建Kafka2.7版本源码运行环境
  • Kafka干货之「零拷贝」
  • 扫盲Kafka？看这一篇就够了！
  • 从一次Kafka宕机说起（JVM hang）
  • 从Kafka中学习高性能系统如何设计

    • 本站QQ群：前端 618073944 | Java 606181507 | Python 626812652 | C/C++ 612253063 | 微信 634508462 | 苹果 692586424 | C#/.net 182808419 | PHP 305140648 | 运维 608723728

    W3xue 的所有内容仅供测试，对任何法律问题及风险不承担任何责任。通过使用本站内容随之而来的风险与本站无关。
    关于我们  |  意见建议  |  捐助我们  |  报错有奖  |  广告合作、友情链接（目前9元/月）请联系QQ：27243702 沸活量
    皖ICP备17017327号-2　皖公网安备34020702000426号