Kafka ¡¾ÈëÃÅһƪÎÄÕ¾͹»ÁË¡¿

³õʶ Kafka

• ʲôÊÇ kafka
  • Kafka ÊÇÓÉ Linkedin ¹«Ë¾¿ª·¢µÄ£¬ËüÊÇÒ»¸ö·Ö²¼Ê½µÄ£¬Ö§³Ö¶à·ÖÇø¡¢¶à¸±±¾£¬»ùÓÚ Zookeeper µÄ·Ö²¼Ê½ÏûÏ¢Á÷ƽ̨£¬ËüͬʱҲÊÇÒ»¿î¿ªÔ´µÄ»ùÓÚ·¢²¼¶©ÔÄģʽµÄÏûÏ¢ÒýÇæϵͳ¡£
• kafka µÄ»ù±¾ÊõÓï
  • ÏûÏ¢£ºKafka ÖеÄÊý¾Ýµ¥Ôª±»³ÆΪÏûÏ¢£¬Ò²±»³ÆΪ¼Ç¼£¬¿ÉÒÔ°ÑËü¿´×÷Êý¾Ý¿â±íÖÐijһÐеļǼ¡¡¡¡
  • Åú´Î£ºÎªÁËÌá¸ßЧÂÊ£¬ ÏûÏ¢»á·ÖÅú´ÎдÈë Kafka£¬Åú´Î¾Í´úÖ¸µÄÊÇÒ»×éÏûÏ¢¡£
  • Ö÷Ì⣺ÏûÏ¢µÄÖÖÀà³ÆΪ Ö÷Ì⣨Topic£©,¿ÉÒÔ˵һ¸öÖ÷Ìâ´ú±íÁËÒ»ÀàÏûÏ¢¡£Ï൱ÓÚÊǶÔÏûÏ¢½øÐзÖÀà¡£Ö÷Ìâ¾ÍÏñÊÇÊý¾Ý¿âÖÐµÄ±í¡£
  • ·ÖÇø£ºÖ÷Ìâ¿ÉÒÔ±»·ÖΪÈô¸É¸ö·ÖÇø£¨partition£©£¬Í¬Ò»¸öÖ÷ÌâÖеķÖÇø¿ÉÒÔ²»ÔÚÒ»¸ö»úÆ÷ÉÏ£¬ÓпÉÄܻᲿÊðÔÚ¶à¸ö»úÆ÷ÉÏ£¬ÓÉ´ËÀ´ÊµÏÖ kafka µÄÉìËõÐÔ£¬µ¥Ò»Ö÷ÌâÖеķÖÇøÓÐÐò£¬µ«ÊÇÎÞ·¨±£Ö¤Ö÷ÌâÖÐËùÓеķÖÇøÓÐÐò

 

• • Éú²úÕߣº ÏòÖ÷Ìâ·¢²¼ÏûÏ¢µÄ¿Í»§¶ËÓ¦ÓóÌÐò³ÆΪÉú²úÕߣ¨Producer£©£¬Éú²úÕßÓÃÓÚ³ÖÐø²»¶ÏµÄÏòij¸öÖ÷Ìâ·¢ËÍÏûÏ¢¡£
  • Ïû·ÑÕߣº¶©ÔÄÖ÷ÌâÏûÏ¢µÄ¿Í»§¶Ë³ÌÐò³ÆΪÏû·ÑÕߣ¨Consumer£©£¬Ïû·ÑÕßÓÃÓÚ´¦ÀíÉú²úÕß²úÉúµÄÏûÏ¢¡£
  • Ïû·ÑÕßȺ×飺Éú²úÕßÓëÏû·ÑÕߵĹØϵ¾ÍÈçͬ²ÍÌüÖеijøʦºÍ¹Ë¿ÍÖ®¼äµÄ¹ØϵһÑù£¬Ò»¸ö³øʦ¶ÔÓ¦¶à¸ö¹Ë¿Í£¬Ò²¾ÍÊÇÒ»¸öÉú²úÕ߶ÔÓ¦¶à¸öÏû·ÑÕߣ¬Ïû·ÑÕßȺ×飨Consumer Group£©Ö¸µÄ¾ÍÊÇÓÉÒ»¸ö»ò¶à¸öÏû·ÑÕß×é³ÉµÄȺÌå¡£ 

• • Æ«ÒÆÁ¿£ºÆ«ÒÆÁ¿£¨Consumer Offset£©ÊÇ365ÌåÓýͶעԪÊý¾Ý£¬ËüÊÇÒ»¸ö²»¶ÏµÝÔöµÄÕûÊýÖµ£¬ÓÃÀ´¼Ç¼Ïû·ÑÕß·¢ÉúÖØƽºâʱµÄλÖã¬ÒÔ±ãÓÃÀ´»Ö¸´Êý¾Ý¡£
  • broker: Ò»¸ö¶ÀÁ¢µÄ Kafka ·þÎñÆ÷¾Í±»³ÆΪ broker£¬broker ½ÓÊÕÀ´×ÔÉú²úÕßµÄÏûÏ¢£¬ÎªÏûÏ¢ÉèÖÃÆ«ÒÆÁ¿£¬²¢Ìá½»ÏûÏ¢µ½´ÅÅ̱£´æ¡£
  • broker ¼¯Èº£ºbroker ÊǼ¯Èº µÄ×é³É²¿·Ö£¬broker ¼¯ÈºÓÉÒ»¸ö»ò¶à¸ö broker ×é³É£¬365ÌåÓýͶע¼¯Èº¶¼ÓÐÒ»¸ö broker ͬʱ³äµ±Á˼¯Èº¿ØÖÆÆ÷µÄ½ÇÉ«£¨×Ô¶¯´Ó¼¯ÈºµÄ»îÔ¾³ÉÔ±ÖÐÑ¡¾Ù³öÀ´£©¡£
  • ¸±±¾£ºKafka ÖÐÏûÏ¢µÄ±¸·ÝÓÖ½Ð×ö ¸±±¾£¨Replica£©£¬¸±±¾µÄÊýÁ¿ÊÇ¿ÉÒÔÅäÖõģ¬Kafka ¶¨ÒåÁËÁ½Àั±¾£ºÁìµ¼Õ߸±±¾£¨Leader Replica£© ºÍ ×·ËæÕ߸±±¾£¨Follower Replica£©£¬Ç°Õ߶ÔÍâÌṩ·þÎñ£¬ºóÕßÖ»ÊDZ»¶¯¸úËæ¡£
  • ÖØƽºâ£ºRebalance¡£Ïû·ÑÕß×éÄÚij¸öÏû·ÑÕßʵÀý¹Òµôºó£¬ÆäËûÏû·ÑÕßʵÀý×Ô¶¯ÖØзÖÅ䶩ÔÄÖ÷Ìâ·ÖÇøµÄ¹ý³Ì¡£Rebalance ÊÇ Kafka Ïû·ÑÕ߶ËʵÏָ߿ÉÓõÄÖØÒªÊֶΡ£
• kafka µÄÌØÐÔ£¨Éè¼ÆÔ­Ôò£©
  • ¸ßÍÌÍ¡¢µÍÑÓ³Ù£ºkakfa ×î´óµÄÌصã¾ÍÊÇÊÕ·¢ÏûÏ¢·Ç³£¿ì£¬kafka ÿÃë¿ÉÒÔ´¦Àí¼¸Ê®ÍòÌõÏûÏ¢£¬ËüµÄ×îµÍÑÓ³ÙÖ»Óм¸ºÁÃë¡£
  • ¸ßÉìËõÐÔ£º 365ÌåÓýͶעÖ÷Ìâ(topic) °üº¬¶à¸ö·ÖÇø(partition)£¬Ö÷ÌâÖеķÖÇø¿ÉÒÔ·Ö²¼ÔÚ²»Í¬µÄÖ÷»ú(broker)ÖС£
  • ³Ö¾ÃÐÔ¡¢¿É¿¿ÐÔ£º Kafka Äܹ»ÔÊÐíÊý¾ÝµÄ³Ö¾Ã»¯´æ´¢£¬ÏûÏ¢±»³Ö¾Ã»¯µ½´ÅÅÌ£¬²¢Ö§³ÖÊý¾Ý±¸·Ý·ÀÖ¹Êý¾Ý¶ªÊ§£¬Kafka µ×²ãµÄÊý¾Ý´æ´¢ÊÇ»ùÓÚ Zookeeper ´æ´¢µÄ£¬Zookeeper 365betÌåÓýÔÚÏßÃÇÖªµÀËüµÄÊý¾ÝÄܹ»³Ö¾Ã´æ´¢¡£
  • ÈÝ´íÐÔ£º ÔÊÐí¼¯ÈºÖеĽڵãʧ°Ü£¬Ä³¸ö½Úµãå´»ú£¬Kafka ¼¯ÈºÄܹ»Õý³£¹¤×÷
  • ¸ß²¢·¢£º Ö§³ÖÊýǧ¸ö¿Í»§¶Ëͬʱ¶Áд
• kafka µÄʹÓó¡¾°
  • »î¶¯¸ú×Ù£ºKafka ¿ÉÒÔÓÃÀ´¸ú×ÙÓû§ÐÐΪ£¬±ÈÈç365betÌåÓýÔÚÏßÃǾ­³£»ØÈ¥ÌÔ±¦¹ºÎÄã´ò¿ªÌÔ±¦µÄÄÇÒ»¿Ì£¬ÄãµÄµÇ½ÐÅÏ¢£¬µÇ½´ÎÊý¶¼»á×÷ΪÏûÏ¢´«Êäµ½ Kafka £¬µ±Äãä¯ÀÀ¹ºÎïµÄʱºò£¬ÄãµÄä¯ÀÀÐÅÏ¢£¬ÄãµÄËÑË÷Ö¸Êý£¬ÄãµÄ¹ºÎï°®ºÃ¶¼»á×÷Ϊһ¸ö¸öÏûÏ¢´«µÝ¸ø Kafka £¬ÕâÑù¾Í¿ÉÒÔÉú³É±¨¸æ£¬¿ÉÒÔ×öÖÇÄÜÍƼö£¬¹ºÂòϲºÃµÈ¡£
  • ´«µÝÏûÏ¢£ºKafka ÁíÍâÒ»¸ö»ù±¾ÓÃ;ÊÇ´«µÝÏûÏ¢£¬Ó¦ÓóÌÐòÏòÓû§·¢ËÍ֪ͨ¾ÍÊÇͨ¹ý´«µÝÏûÏ¢À´ÊµÏֵģ¬ÕâЩӦÓÃ×é¼þ¿ÉÒÔÉú³ÉÏûÏ¢£¬¶ø²»ÐèÒª¹ØÐÄÏûÏ¢µÄ¸ñʽ£¬Ò²²»ÐèÒª¹ØÐÄÏûÏ¢ÊÇÈçºÎ·¢Ë͵ġ£
  • ¶ÈÁ¿Ö¸±ê£ºKafkaÒ²¾­³£ÓÃÀ´¼Ç¼ÔËÓª¼à¿ØÊý¾Ý¡£°üÀ¨ÊÕ¼¯¸÷ÖÖ·Ö²¼Ê½Ó¦ÓõÄÊý¾Ý£¬Éú²ú¸÷ÖÖ²Ù×÷µÄ¼¯Öз´À¡£¬±ÈÈ籨¾¯ºÍ±¨¸æ¡£
  • ÈÕÖ¾¼Ç¼£ºKafka µÄ»ù±¾¸ÅÄîÀ´Ô´ÓÚÌá½»ÈÕÖ¾£¬±ÈÈç365betÌåÓýÔÚÏßÃÇ¿ÉÒÔ°ÑÊý¾Ý¿âµÄ¸üз¢Ë͵½ Kafka ÉÏ£¬ÓÃÀ´¼Ç¼Êý¾Ý¿âµÄ¸üÐÂʱ¼ä£¬Í¨¹ýkafkaÒÔͳһ½Ó¿Ú·þÎñµÄ·½Ê½¿ª·Å¸ø¸÷ÖÖconsumer£¬ÀýÈçhadoop¡¢Hbase¡¢SolrµÈ¡£
  • Á÷ʽ´¦Àí£ºÁ÷ʽ´¦ÀíÊÇÓÐÒ»¸öÄܹ»Ìṩ¶àÖÖÓ¦ÓóÌÐòµÄÁìÓò¡£
  • ÏÞÁ÷Ï÷·å£ºKafka ¶àÓÃÓÚ»¥ÁªÍøÁìÓòijһʱ¿ÌÇëÇóÌرð¶àµÄÇé¿öÏ£¬¿ÉÒÔ°ÑÇëÇóдÈëKafka ÖУ¬±ÜÃâÖ±½ÓÇëÇóºó¶Ë³ÌÐòµ¼Ö·þÎñ±ÀÀ£¡£
• kafka µÄÏûÏ¢¶ÓÁÐ
  • Kafka µÄÏûÏ¢¶ÓÁÐÒ»°ã·ÖΪÁ½ÖÖģʽ£ºµã¶ÔµãģʽºÍ·¢²¼¶©ÔÄģʽ
  • Kafka ÊÇÖ§³ÖÏû·ÑÕßȺ×éµÄ£¬Ò²¾ÍÊÇ˵ Kafka ÖлáÓÐÒ»¸ö»òÕ߶à¸öÏû·ÑÕߣ¬Èç¹ûÒ»¸öÉú²úÕßÉú²úµÄÏûÏ¢ÓÉÒ»¸öÏû·ÑÕß½øÐÐÏû·ÑµÄ»°£¬ÄÇôÕâÖÖģʽ¾ÍÊǵã¶Ôµãģʽ

• • Èç¹ûÒ»¸öÉú²úÕß»òÕ߶à¸öÉú²úÕß²úÉúµÄÏûÏ¢Äܹ»±»¶à¸öÏû·ÑÕßͬʱÏû·ÑµÄÇé¿ö£¬ÕâÑùµÄÏûÏ¢¶ÓÁгÉΪ·¢²¼¶©ÔÄģʽµÄÏûÏ¢¶ÓÁÐ

• kafka ϵͳ¼Ü¹¹

• • ÈçÉÏͼËùʾ£¬Ò»¸öµäÐ굀 Kafka ¼¯ÈºÖаüº¬Èô¸ÉProducer£¨¿ÉÒÔÊÇwebÇ°¶Ë²úÉúµÄPage View£¬»òÕßÊÇ·þÎñÆ÷ÈÕÖ¾£¬ÏµÍ³CPU¡¢MemoryµÈ£©£¬Èô¸Ébroker£¨KafkaÖ§³ÖˮƽÀ©Õ¹£¬Ò»°ãbrokerÊýÁ¿Ô½¶à£¬¼¯ÈºÍÌÍÂÂÊÔ½¸ß£©£¬Èô¸ÉConsumer Group£¬ÒÔ¼°Ò»¸öZookeeper¼¯Èº¡£Kafkaͨ¹ýZookeeper¹ÜÀí¼¯ÈºÅäÖã¬Ñ¡¾Ùleader£¬ÒÔ¼°ÔÚConsumer Group·¢Éú±ä»¯Ê±½øÐÐrebalance¡£ProducerʹÓÃpushģʽ½«ÏûÏ¢·¢²¼µ½broker£¬ConsumerʹÓÃpullģʽ´Óbroker¶©ÔIJ¢Ïû·ÑÏûÏ¢¡£
• ºËÐÄ API
  • kafka ÓÐËĸöºËÐÄAPI£¬·Ö±ðÊÇ
    • Producer API£¬ËüÔÊÐíÓ¦ÓóÌÐòÏòÒ»¸ö»ò¶à¸ö topics ÉÏ·¢ËÍÏûÏ¢¼Ç¼
    • Consumer API£¬ÔÊÐíÓ¦ÓóÌÐò¶©ÔÄÒ»¸ö»ò¶à¸ö topics ²¢´¦ÀíΪÆäÉú³ÉµÄ¼Ç¼Á÷
    • Streams API£¬ËüÔÊÐíÓ¦ÓóÌÐò×÷ΪÁ÷´¦ÀíÆ÷£¬´ÓÒ»¸ö»ò¶à¸öÖ÷ÌâÖÐÏû·ÑÊäÈëÁ÷²¢ÎªÆäÉú³ÉÊä³öÁ÷£¬ÓÐЧµÄ½«ÊäÈëÁ÷ת»»ÎªÊä³öÁ÷¡£
    • Connector API£¬ËüÔÊÐí¹¹½¨ºÍÔËÐн« Kafka Ö÷ÌâÁ¬½Óµ½ÏÖÓÐÓ¦ÓóÌÐò»òÊý¾ÝϵͳµÄ¿ÉÓÃÉú²úÕߺÍÏû·ÑÕß¡£ÀýÈ磬¹ØϵÊý¾Ý¿âµÄÁ¬½ÓÆ÷¿ÉÄܻᲶ»ñ¶Ô±íµÄËùÓиü¸Ä

• kafka ΪºÎÈç´ËÖ®¿ì
  • Kafka ʵÏÖÁËÁ㿽±´Ô­ÀíÀ´¿ìËÙÒƶ¯Êý¾Ý£¬±ÜÃâÁËÄÚºËÖ®¼äµÄÇл»¡£Kafka ¿ÉÒÔ½«Êý¾Ý¼Ç¼·ÖÅú·¢ËÍ£¬´ÓÉú²úÕßµ½Îļþϵͳ£¨Kafka Ö÷ÌâÈÕÖ¾£©µ½Ïû·ÑÕߣ¬¿ÉÒԶ˵½¶ËµÄ²é¿´ÕâЩÅú´ÎµÄÊý¾Ý¡£
  • Åú´¦ÀíÄܹ»½øÐиüÓÐЧµÄÊý¾ÝѹËõ²¢¼õÉÙ I/O ÑÓ³Ù£¬Kafka ²Éȡ˳ÐòдÈë´ÅÅ̵ķ½Ê½£¬±ÜÃâÁËËæ»ú´ÅÅÌÑ°Ö·µÄÀË·Ñ£¬¸ü¶à¹ØÓÚ´ÅÅÌÑ°Ö·µÄÁ˽⣬Çë²ÎÔÄ ³ÌÐòÔ±ÐèÒªÁ˽âµÄÓ²ºË֪ʶ֮´ÅÅÌ 
• ×ܽáÒ»ÏÂÆäʵ¾ÍÊÇËĸöÒªµã
  • ˳Ðò¶Áд
  • Á㿽±´
  • ÏûϢѹËõ
  • ·ÖÅú·¢ËÍ

 

Kafka °²×°ºÍÖØÒªÅäÖÃ

1. Kafka °²×°365betÌåÓýÔÚÏßÔÚ Kafka ϵÁеÚһƪӦ¸Ã±È½ÏÏêϸÁË£¬ÏêÇé¼û´øÄãÕÇ×ËÊƵÄÈÏʶһÏÂkafka ÕâƪÎÄÕ¡£
2. ÄÇ365betÌåÓýÔÚÏßÃÇ»¹ÊÇÖ÷ÒªÀ´ËµÒ»Ï Kafka ÖеÄÖØÒª²ÎÊýÅäÖðɣ¬ÕâЩ²ÎÊý¶Ô Kafka À´ËµÊǷdz£ÖØÒªµÄ¡£

• broker ¶ËÅäÖÃ
  • broker.id
    • 365ÌåÓýͶע kafka broker ¶¼ÓÐÒ»¸öΨһµÄ±êʶÀ´365ÌåÓýͶע£¬Õâ¸öΨһµÄ±êʶ·û¼´ÊÇ broker.id£¬ËüµÄĬÈÏÖµÊÇ 0¡£Õâ¸öÖµÔÚ kafka ¼¯ÈºÖбØÐëÊÇΨһµÄ£¬Õâ¸öÖµ¿ÉÒÔÈ365betÌåÓýÔÚÏßâÉ趨£¬
  • port
    • Èç¹ûʹÓÃÅäÖÃÑù±¾À´Æô¶¯ kafka£¬Ëü»á¼àÌý 9092 ¶Ë¿Ú¡£ÐÞ¸Ä port ÅäÖòÎÊý¿ÉÒÔ°ÑËüÉèÖóÉÈ365betÌåÓýÔÚÏßâµÄ¶Ë¿Ú¡£Òª×¢Ò⣬Èç¹ûʹÓà 1024 ÒÔϵĶ˿ڣ¬ÐèҪʹÓà root ȨÏÞÆô¶¯ kakfa¡£
  • zookeeper.connect
    • ÓÃÓÚ±£´æ broker ÔªÊý¾ÝµÄ Zookeeper µØÖ·ÊÇͨ¹ý zookeeper.connect À´Ö¸¶¨µÄ¡£±ÈÈç365betÌåÓýÔÚÏß¿ÉÒÔÕâôָ¶¨ localhost:2181 365ÌåÓýͶעÕâ¸ö Zookeeper ÊÇÔËÐÐÔÚ±¾µØ 2181 ¶Ë¿ÚÉϵġ£365betÌåÓýÔÚÏßÃÇÒ²¿ÉÒÔͨ¹ý ±ÈÈç365betÌåÓýÔÚÏßÃÇ¿ÉÒÔͨ¹ý zk1:2181,zk2:2181,zk3:2181 À´Ö¸¶¨ zookeeper.connect µÄ¶à¸ö²ÎÊýÖµ¡£¸ÃÅäÖòÎÊýÊÇÓÃðºÅ·Ö¸îµÄÒ»×é hostname:port/path ÁÐ±í£¬Æ京ÒåÈçÏÂ
    • hostname ÊÇ Zookeeper ·þÎñÆ÷µÄ»úÆ÷Ãû»òÕß ip µØÖ·¡£
    • port ÊÇ Zookeeper ¿Í»§¶ËµÄ¶Ë¿ÚºÅ
    • /path ÊÇ¿ÉÑ¡ÔñµÄ Zookeeper ·¾¶£¬Kafka ·¾¶ÊÇʹÓÃÁË chroot »·¾³£¬Èç¹û²»Ö¸¶¨Ä¬ÈÏʹÓøú·¾¶¡£
    • Èç¹ûÄãÓÐÁ½Ì× Kafka ¼¯Èº£¬¼ÙÉè·Ö±ð½ÐËüÃÇ kafka1 ºÍ kafka2£¬ÄÇôÁ½Ì×¼¯ÈºµÄzookeeper.connect²ÎÊý¿ÉÒÔÕâÑùÖ¸¶¨£ºzk1:2181,zk2:2181,zk3:2181/kafka1ºÍzk1:2181,zk2:2181,zk3:2181/kafka2
  • log.dirs
    • Kafka °ÑËùÓеÄÏûÏ¢¶¼±£´æµ½´ÅÅÌÉÏ£¬´æ·ÅÕâЩÈÕ־Ƭ¶ÎµÄĿ¼ÊÇͨ¹ý log.dirs À´Öƶ¨µÄ£¬ËüÊÇÓÃÒ»×鶺ºÅÀ´·Ö¸îµÄ±¾µØϵͳ·¾¶£¬log.dirs ÊÇûÓÐĬÈÏÖµµÄ£¬Äã±ØÐëÊÖ¶¯Ö¸¶¨ËûµÄĬÈÏÖµ¡£Æäʵ»¹ÓÐÒ»¸ö²ÎÊýÊÇ log.dir£¬ÈçÄãËùÖª£¬Õâ¸öÅäÖÃÊÇûÓÐ s µÄ£¬Ä¬ÈÏÇé¿öÏÂÖ»ÓÃÅäÖà log.dirs ¾ÍºÃÁË£¬±ÈÈçÄã¿ÉÒÔͨ¹ý/home/kafka1,/home/kafka2,/home/kafka3 ÕâÑùÀ´ÅäÖÃÕâ¸ö²ÎÊýµÄÖµ¡£
  • num.recovery.threads.per.data.dir
    • ¶ÔÓÚÈçÏÂ3ÖÖÇé¿ö£¬Kafka »áʹÓÿÉÅäÖõÄÏ̳߳ØÀ´´¦ÀíÈÕ־Ƭ¶Î¡£
    • ·þÎñÆ÷Õý³£Æô¶¯£¬ÓÃÓÚ´ò¿ª365ÌåÓýͶע·ÖÇøµÄÈÕ־Ƭ¶Î£»
    • ·þÎñÆ÷±ÀÀ£ºóÖØÆô£¬ÓÃÓÚ¼ì²éºÍ½Ø¶Ï365ÌåÓýͶע·ÖÇøµÄÈÕ־Ƭ¶Î£»
    • ·þÎñÆ÷Õý³£¹Ø±Õ£¬ÓÃÓڹرÕÈÕ־Ƭ¶Î¡£
    • ĬÈÏÇé¿öÏ£¬365ÌåÓýͶעÈÕ־Ŀ¼ֻʹÓÃÒ»¸öÏ̡߳£ÒòΪÕâЩÏß³ÌÖ»ÊÇÔÚ·þÎñÆ÷Æô¶¯ºÍ¹Ø±Õʱ»áÓõ½£¬365betÌåÓýÔÚÏßÍêÈ«¿ÉÒÔÉèÖôóÁ¿µÄÏß³ÌÀ´´ïµ½¾®ÐвÙ×÷µÄÄ¿µÄ¡£ÌرðÊǶÔÓÚ°üº¬´óÁ¿·ÖÇøµÄ·þÎñÆ÷À´Ëµ£¬Ò»µ©·¢Éú±À·ß£¬ÔÚ½øÐлָ´Ê±Ê¹Óþ®ÐвÙ×÷¿ÉÄÜ»áÊ¡ÏÂÊýСʱµÄʱ¼ä¡£ÉèÖô˲ÎÊýʱÐèҪעÒ⣬ËùÅäÖõÄÊý×Ö¶ÔÓ¦µÄÊÇ log.dirs Ö¸¶¨µÄµ¥¸öÈÕ־Ŀ¼¡£Ò²¾ÍÊÇ˵£¬Èç¹û num.recovery.threads.per.data.dir ±»ÉèΪ 8£¬²¢ÇÒ log.dir Ö¸¶¨ÁË 3 ¸ö·¾¶£¬ÄÇô×ܹ²ÐèÒª 24 ¸öÏ̡߳£
  • auto.create.topics.enable
    • ĬÈÏÇé¿öÏ£¬kafka »áʹÓÃÈýÖÖ·½Ê½À´×Ô¶¯´´½¨Ö÷Ì⣬ÏÂÃæÊÇÈýÖÖÇé¿ö£º
    • µ±Ò»¸öÉú²úÕß¿ªÊ¼ÍùÖ÷ÌâдÈëÏûϢʱ
    • µ±Ò»¸öÏû·ÑÕß¿ªÊ¼´ÓÖ÷Ìâ¶ÁÈ¡ÏûϢʱ
    • µ±È365betÌåÓýÔÚÏßâÒ»¸ö¿Í»§¶ËÏòÖ÷Ìâ·¢ËÍÔªÊý¾ÝÇëÇóʱ
    • auto.create.topics.enable²ÎÊý365betÌåÓýÔÚÏß½¨Òé×îºÃÉèÖÃ³É false£¬¼´²»ÔÊÐí×Ô¶¯´´½¨ Topic¡£ÔÚ365betÌåÓýÔÚÏßÃǵÄÏßÉÏ»·¾³ÀïÃæÓкܶàÃû×ÖÏ¡Ææ¹Å¹ÖµÄ Topic£¬365betÌåÓýÔÚÏßÏë´ó¸Å¶¼ÊÇÒòΪ¸Ã²ÎÊý±»ÉèÖóÉÁË true µÄÔµ¹Ê¡£
• Ö÷ÌâĬÈÏÅäÖÃ
  • Kafka Ϊд´½¨µÄÖ÷ÌâÌṩÁ˺ܶàĬÈÏÅäÖòÎÊý£¬ÏÂÃæ¾ÍÀ´Ò»ÆðÈÏʶһÏÂÕâЩ
  • num.partitions
    • num.partitions ²ÎÊýÖ¸¶¨ÁËд´½¨µÄÖ÷ÌâÐèÒª°üº¬¶àÉÙ¸ö·ÖÇø¡£Èç¹ûÆôÓÃÁËÖ÷Ìâ×Ô¶¯´´½¨¹¦ÄÜ£¨¸Ã¹¦ÄÜÊÇĬÈÏÆôÓõģ©£¬Ö÷Ìâ·ÖÇøµÄ¸öÊý¾ÍÊǸòÎÊýÖ¸¶¨µÄÖµ¡£¸Ã²ÎÊýµÄĬÈÏÖµÊÇ 1¡£Òª×¢Ò⣬365betÌåÓýÔÚÏßÃÇ¿ÉÒÔÔö¼ÓÖ÷Ìâ·ÖÇøµÄ¸öÊý£¬µ«²»ÄܼõÉÙ·ÖÇøµÄ¸öÊý¡£
  • default.replication.factor
    • Õâ¸ö²ÎÊý±È½Ï¼òµ¥£¬Ëü365ÌåÓýͶע kafka±£´æÏûÏ¢µÄ¸±±¾Êý£¬Èç¹ûÒ»¸ö¸±±¾Ê§Ð§ÁË£¬ÁíÒ»¸ö»¹¿ÉÒÔ¼ÌÐøÌṩ·þÎñdefault.replication.factor µÄĬÈÏֵΪ1£¬Õâ¸ö²ÎÊýÔÚÄãÆôÓÃÁËÖ÷Ìâ×Ô¶¯´´½¨¹¦ÄܺóÓÐЧ¡£
  • log.retention.ms
    • Kafka ͨ³£¸ù¾Ýʱ¼äÀ´¾ö¶¨Êý¾Ý¿ÉÒÔ±£Áô¶à¾Ã¡£Ä¬ÈÏʹÓà log.retention.hours ²ÎÊýÀ´ÅäÖÃʱ¼ä£¬Ä¬ÈÏÊÇ 168 ¸öСʱ£¬Ò²¾ÍÊÇÒ»ÖÜ¡£³ý´ËÖ®Í⣬»¹ÓÐÁ½¸ö²ÎÊý log.retention.minutes ºÍ log.retentiion.ms ¡£ÕâÈý¸ö²ÎÊý×÷ÓÃÊÇÒ»ÑùµÄ£¬¶¼ÊǾö¶¨ÏûÏ¢¶à¾ÃÒÔºó±»É¾³ý£¬ÍƼöʹÓà log.retention.ms¡£
  • log.retention.bytes
    • Áí365ÌåÓýͶע±£ÁôÏûÏ¢µÄ·½Ê½ÊÇÅжÏÏûÏ¢ÊÇ·ñ¹ýÆÚ¡£ËüµÄֵͨ¹ý²ÎÊý log.retention.bytes À´Ö¸¶¨£¬×÷ÓÃÔÚÿһ¸ö·ÖÇøÉÏ¡£Ò²¾ÍÊÇ˵£¬Èç¹ûÓÐÒ»¸ö°üº¬ 8 ¸ö·ÖÇøµÄÖ÷Ì⣬²¢ÇÒ log.retention.bytes ±»ÉèÖÃΪ 1GB£¬ÄÇôÕâ¸öÖ÷Ìâ×î¶à¿ÉÒÔ±£Áô 8GB Êý¾Ý¡£365betÌåÓýÔÚÏߣ¬µ±Ö÷ÌâµÄ·ÖÇø¸öÊýÔö¼Óʱ£¬Õû¸öÖ÷Ìâ¿ÉÒÔ±£ÁôµÄÊý¾ÝÒ²ËæÖ®Ôö¼Ó¡£
  • log.segment.bytes
    • ÉÏÊöµÄÈÕÖ¾¶¼ÊÇ×÷ÓÃÔÚÈÕ־Ƭ¶ÎÉÏ£¬¶ø²»ÊÇ×÷ÓÃÔÚµ¥¸öÏûÏ¢ÉÏ¡£µ±ÏûÏ¢µ½´ï broker ʱ£¬ËüÃDZ»×·¼Óµ½·ÖÇøµÄµ±Ç°ÈÕ־Ƭ¶ÎÉÏ£¬µ±ÈÕ־Ƭ¶Î´óСµ½´ï log.segment.bytes Ö¸¶¨ÉÏÏÞ£¨Ä¬ÈÏΪ 1GB£©Ê±£¬µ±Ç°ÈÕ־Ƭ¶Î¾Í»á±»¹Ø±Õ£¬Ò»¸öеÄÈÕ־Ƭ¶Î±»´ò¿ª¡£Èç¹ûÒ»¸öÈÕ־Ƭ¶Î±»¹Ø±Õ£¬¾Í¿ªÊ¼µÈ´ý¹ýÆÚ¡£Õâ¸ö²ÎÊýµÄֵԽС£¬¾ÍÔ½»áƵ·±µÄ¹Ø±ÕºÍ·ÖÅäÐÂÎļþ£¬´Ó¶ø½µµÍ´ÅÅÌдÈëµÄÕûÌåЧÂÊ¡£
  • log.segment.ms
    • ÉÏÃæÌáµ½ÈÕ־Ƭ¶Î¾­¹Ø±ÕºóÐèµÈ´ý¹ýÆÚ£¬ÄÇô log.segment.ms Õâ¸ö²ÎÊý¾ÍÊÇÖ¸¶¨ÈÕÖ¾¶à³¤Ê±¼ä±»¹Ø±ÕµÄ²ÎÊýºÍ£¬log.segment.ms ºÍ log.retention.bytes Ò²²»´æÔÚ»¥³âÎÊÌâ¡£ÈÕ־Ƭ¶Î»áÔÚ´óС»òʱ¼äµ½´ïÉÏÏÞʱ±»¹Ø±Õ£¬¾Í¿´ÄĸöÌõ¼þÏȵõ½Âú×ã¡£
  • message.max.bytes
    • broker ͨ¹ýÉèÖà message.max.bytes ²ÎÊýÀ´ÏÞÖƵ¥¸öÏûÏ¢µÄ´óС£¬Ä¬ÈÏÊÇ 1000 000£¬ Ò²¾ÍÊÇ 1MB£¬Èç¹ûÉú²úÕß³¢ÊÔ·¢Ë͵ÄÏûÏ¢³¬¹ýÕâ¸ö´óС£¬²»½öÏûÏ¢²»»á±»½ÓÊÕ£¬»¹»áÊÕµ½ broker ·µ»ØµÄ´íÎóÏûÏ¢¡£¸úÆäËûÓë×Ö½ÚÏà¹ØµÄÅäÖòÎÊýÒ»Ñù£¬¸Ã²ÎÊýÖ¸µÄÊÇѹËõºóµÄÏûÏ¢´óС£¬Ò²¾ÍÊÇ˵£¬Ö»ÒªÑ¹ËõºóµÄÏûϢСÓÚ mesage.max.bytes£¬ÄÇôÏûÏ¢µÄʵ¼Ê´óС¿ÉÒÔ´óÓÚÕâ¸öÖµ
    • Õâ¸öÖµ¶ÔÐÔÄÜÓÐÏÔÖøµÄÓ°Ïì¡£ÖµÔ½´ó£¬ÄÇô¸ºÔð´¦ÀíÍøÂçÁ¬½ÓºÍÇëÇóµÄÏ߳̾ÍÐèÒª»¨Ô½¶àµÄʱ¼äÀ´´¦ÀíÕâЩÇëÇó¡£Ëü»¹»áÔö¼Ó´ÅÅÌдÈë¿éµÄ´óС£¬´Ó¶øÓ°Ïì IO ÍÌÍÂÁ¿¡£
  • retention.ms
    • ¹æ¶¨Á˸ÃÖ÷ÌâÏûÏ¢±»±£´æµÄʱ³££¬Ä¬ÈÏÊÇ7Ì죬¼´¸ÃÖ÷ÌâÖ»Äܱ£´æ7ÌìµÄÏûÏ¢£¬Ò»µ©ÉèÖÃÁËÕâ¸öÖµ£¬Ëü»á¸²¸Çµô Broker ¶ËµÄÈ«¾Ö²ÎÊýÖµ¡£
  • retention.bytes
    • retention.bytes£º¹æ¶¨ÁËҪΪ¸Ã Topic Ô¤Áô¶à´óµÄ´ÅÅ̿ռ䡣ºÍÈ«¾Ö²ÎÊý×÷ÓÃÏàËÆ£¬Õâ¸öֵͨ³£ÔÚ¶à×⻧µÄ Kafka ¼¯ÈºÖлáÓÐÓÃÎäÖ®µØ¡£µ±Ç°Ä¬ÈÏÖµÊÇ -1£¬365ÌåÓýͶע¿ÉÒÔÎÞÏÞʹÓôÅÅ̿ռ䡣

• JVM ²ÎÊýÅäÖÃ
  • JDK °æ±¾Ò»°ãÍƼöÖ±½ÓʹÓà JDK1.8£¬Õâ¸ö°æ±¾Ò²ÊÇÏÖÔÚÖйú´ó²¿·Ö³ÌÐòÔ±µÄÊ×Ñ¡°æ±¾¡£
  • ˵µ½ JVM ¶ËÉèÖ㬾ÍÈƲ»¿ª¶ÑÕâ¸ö»°Ì⣬ҵ½ç×îÍƳçµÄ365ÌåÓýͶעÉèÖ÷½Ê½¾ÍÊÇÖ±½Ó½« JVM ¶Ñ´óСÉèÖÃΪ 6GB£¬ÕâÑù»á±ÜÃâºÜ¶à Bug ³öÏÖ¡£
  • JVM ¶ËÅäÖõÄÁíÒ»¸öÖØÒª²ÎÊý¾ÍÊÇÀ¬»ø»ØÊÕÆ÷µÄÉèÖã¬Ò²¾ÍÊÇƽʱ³£ËµµÄ GC ÉèÖá£Èç¹ûÄãÒÀÈ»ÔÚʹÓà Java 7£¬ÄÇô¿ÉÒÔ¸ù¾ÝÒÔÏ·¨ÔòÑ¡ÔñºÏÊʵÄÀ¬»ø»ØÊÕÆ÷£º
    • Èç¹û Broker ËùÔÚ»úÆ÷µÄ CPU ×ÊÔ´·Ç³£³äÔ££¬½¨ÒéʹÓà CMS ÊÕ¼¯Æ÷¡£ÆôÓ÷½·¨ÊÇÖ¸¶¨-XX:+UseCurrentMarkSweepGC¡£
    • ·ñÔò£¬Ê¹ÓÃÍÌÍÂÁ¿ÊÕ¼¯Æ÷¡£¿ªÆô·½·¨ÊÇÖ¸¶¨-XX:+UseParallelGC¡£
  • µ±È»ÁË£¬Èç¹ûÄãÒѾ­ÔÚʹÓà Java 8 ÁË£¬ÄÇô¾ÍÓÃĬÈ쵀 G1 ÊÕ¼¯Æ÷¾ÍºÃÁË¡£ÔÚûÓÐÈκε÷ÓŵÄÇé¿öÏ£¬G1 ±íÏÖµÃÒª±È CMS ³öÉ«£¬Ö÷ÒªÌåÏÖÔÚ¸üÉÙµÄ Full GC£¬ÐèÒªµ÷ÕûµÄ²ÎÊý¸üÉٵȣ¬365betÌåÓýÔÚÏßʹÓà G1 ¾ÍºÃÁË¡£
  • Ò»°ã G1 µÄµ÷ÕûÖ»ÐèÒªÕâÁ½¸ö²ÎÊý¼´¿É
  • MaxGCPauseMillis
    • ¸Ã²ÎÊýÖ¸¶¨Ã¿´ÎÀ¬»ø»ØÊÕĬÈϵÄÍ£¶Ùʱ¼ä¡£¸ÃÖµ²»Êǹ̶¨µÄ£¬G1¿ÉÒÔ¸ù¾ÝÐèҪʹÓøü³¤µÄʱ¼ä¡£ËüµÄĬÈÏÖµÊÇ 200ms£¬Ò²¾ÍÊÇ˵£¬Ã¿Ò»ÂÖÀ¬»ø»ØÊÕ´ó¸ÅÐèÒª200 ms µÄʱ¼ä¡£
  • InitiatingHeapOccupancyPercent
    • ¸Ã²ÎÊýÖ¸¶¨ÁË G1 Æô¶¯ÐÂÒ»ÂÖÀ¬»ø»ØÊÕ֮ǰ¿ÉÒÔʹÓõĶÑÄÚ´æ°Ù·Ö±È£¬Ä¬ÈÏÖµÊÇ45£¬Õâ¾Í±íÃ÷G1ÔÚ¶ÑʹÓÃÂʵ½´ï45֮ǰ²»»áÆôÓÃÀ¬»ø»ØÊÕ¡£Õâ¸ö°Ù·Ö±È°üÀ¨ÐÂÉú´úºÍÀÏÄê´ú¡£

 

Kafka Producer

• ÔÚ Kafka ÖУ¬365betÌåÓýÔÚÏßÃǰѲúÉúÏûÏ¢µÄÄÇ365ÌåÓýͶע³ÆΪÉú²úÕߣ¬±ÈÈç365betÌåÓýÔÚÏßÃǾ­³£»ØÈ¥ÌÔ±¦¹ºÎÄã´ò¿ªÌÔ±¦µÄÄÇÒ»¿Ì£¬ÄãµÄµÇ½ÐÅÏ¢£¬µÇ½´ÎÊý¶¼»á×÷ΪÏûÏ¢´«Êäµ½ Kafka ºǫ́£¬µ±Äãä¯ÀÀ¹ºÎïµÄʱºò£¬ÄãµÄä¯ÀÀÐÅÏ¢£¬ÄãµÄËÑË÷Ö¸Êý£¬ÄãµÄ¹ºÎï°®ºÃ¶¼»á×÷Ϊһ¸ö¸öÏûÏ¢´«µÝ¸ø Kafka ºǫ́£¬È»ºóÌÔ±¦»á¸ù¾ÝÄãµÄ°®ºÃ×öÖÇÄÜÍƼö£¬ÖÂʹÄãµÄÇ®°ü´ÓÀ´¶¼½û²»×¡ÓÕ»ó£¬ÄÇôÕâЩÉú²úÕß²úÉúµÄÏûÏ¢ÊÇÔõô´«µ½ Kafka Ó¦ÓóÌÐòµÄÄØ£¿·¢Ë͹ý³ÌÊÇÔõôÑùµÄÄØ£¿
• ¾¡¹ÜÏûÏ¢µÄ²úÉú·Ç³£¼òµ¥£¬µ«ÊÇÏûÏ¢µÄ·¢Ë͹ý³Ì»¹ÊDZȽϸ´Ôӵģ¬Èçͼ

• 365betÌåÓýÔÚÏßÃÇ´Ó´´½¨Ò»¸öProducerRecord ¶ÔÏó¿ªÊ¼£¬ProducerRecord ÊÇ Kafka ÖеÄÒ»¸öºËÐÄÀ࣬Ëü´ú±íÁËÒ»×é Kafka ÐèÒª·¢Ë굀 key/value ¼üÖµ¶Ô£¬ËüÓɼǼҪ·¢Ë͵½µÄÖ÷ÌâÃû³Æ£¨Topic Name£©£¬¿ÉÑ¡µÄ·ÖÇøºÅ£¨Partition Number£©ÒÔ¼°¿ÉÑ¡µÄ¼üÖµ¶Ô¹¹³É¡£
• ÔÚ·¢ËÍ ProducerRecord ʱ£¬365betÌåÓýÔÚÏßÃÇÐèÒª½«¼üÖµ¶Ô¶ÔÏóÓÉÐòÁл¯Æ÷ת»»Îª×Ö½ÚÊý×飬ÕâÑùËüÃDzÅÄܹ»ÔÚÍøÂçÉÏ´«Ê䡣ȻºóÏûÏ¢µ½´ïÁË·ÖÇøÆ÷¡£
• Èç¹û·¢Ë͹ý³ÌÖÐÖ¸¶¨ÁËÓÐЧµÄ·ÖÇøºÅ£¬ÄÇôÔÚ·¢ËͼǼʱ½«Ê¹Óø÷ÖÇø¡£Èç¹û·¢Ë͹ý³ÌÖÐδָ¶¨·ÖÇø£¬Ôò½«Ê¹ÓÃkey µÄ hash º¯ÊýÓ³ÉäÖ¸¶¨Ò»¸ö·ÖÇø¡£Èç¹û·¢Ë͵Ĺý³ÌÖмÈûÓзÖÇøºÅҲûÓУ¬Ôò½«ÒÔÑ­»·µÄ·½Ê½·ÖÅäÒ»¸ö·ÖÇø¡£Ñ¡ºÃ·ÖÇøºó£¬Éú²úÕß¾ÍÖªµÀÏòÄĸöÖ÷ÌâºÍ·ÖÇø·¢ËÍÊý¾ÝÁË¡£
• ProducerRecord »¹ÓйØÁªµÄʱ¼ä´Á£¬Èç¹ûÓû§Ã»ÓÐÌṩʱ¼ä´Á£¬ÄÇôÉú²úÕß½«»áÔڼǼÖÐʹÓõ±Ç°µÄʱ¼ä×÷Ϊʱ¼ä´Á¡£Kafka ×îÖÕʹÓõÄʱ¼ä´ÁÈ¡¾öÓÚ topic Ö÷ÌâÅäÖõÄʱ¼ä´ÁÀàÐÍ¡£
  • Èç¹û½«Ö÷ÌâÅäÖÃΪʹÓà CreateTime£¬ÔòÉú²úÕ߼ǼÖеÄʱ¼ä´Á½«ÓÉ broker ʹÓá£
  • Èç¹û½«Ö÷ÌâÅäÖÃΪʹÓÃLogAppendTime£¬ÔòÉú²úÕ߼ǼÖеÄʱ¼ä´ÁÔÚ½«ÏûÏ¢Ìí¼Óµ½ÆäÈÕÖ¾ÖÐʱ£¬½«ÓÉ broker ÖØд¡£
• È»ºó£¬ÕâÌõÏûÏ¢±»´æ·ÅÔÚÒ»¸ö¼Ç¼Åú´ÎÀÕâ¸öÅú´ÎÀïµÄËùÓÐÏûÏ¢»á±»·¢Ë͵½ÏàͬµÄÖ÷ÌâºÍ·ÖÇøÉÏ¡£ÓÉÒ»¸ö¶ÀÁ¢µÄÏ̸߳ºÔð°ÑËüÃÇ·¢µ½ Kafka Broker ÉÏ¡£
• Kafka Broker ÔÚÊÕµ½ÏûϢʱ»á·µ»ØÒ»¸öÏìÓ¦£¬Èç¹ûдÈë³É¹¦£¬»á·µ»ØÒ»¸ö RecordMetaData ¶ÔÏó£¬Ëü°üº¬ÁËÖ÷ÌâºÍ·ÖÇøÐÅÏ¢£¬ÒÔ¼°¼Ç¼ÔÚ·ÖÇøÀïµÄÆ«ÒÆÁ¿£¬ÉÏÃæÁ½ÖÖµÄʱ¼ä´ÁÀàÐÍÒ²»á·µ»Ø¸øÓû§¡£Èç¹ûдÈëʧ°Ü£¬»á·µ»ØÒ»¸ö´íÎó¡£Éú²úÕßÔÚÊÕµ½´íÎóÖ®ºó»á³¢ÊÔÖØз¢ËÍÏûÏ¢£¬¼¸´ÎÖ®ºóÈç¹û»¹ÊÇʧ°ÜµÄ»°£¬¾Í·µ»Ø´íÎóÏûÏ¢¡£ 

• ´´½¨ kafka Éú²úÕß
  • ÒªÏò Kafka дÈëÏûÏ¢£¬Ê×ÏÈÐèÒª´´½¨Ò»¸öÉú²úÕ߶ÔÏ󣬲¢ÉèÖÃһЩÊôÐÔ¡£Kafka Éú²úÕßÓÐ3¸ö±ØÑ¡µÄÊôÐÔ
  • bootstrap.servers
    • ¸ÃÊôÐÔÖ¸¶¨ broker µÄµØÖ·Çåµ¥£¬µØÖ·µÄ¸ñʽΪ host:port¡£Çåµ¥Àï²»ÐèÒª°üº¬ËùÓÐµÄ broker µØÖ·£¬Éú²úÕß»á´Ó¸ø¶¨µÄ broker Àï²éÕÒµ½ÆäËûµÄ broker ÐÅÏ¢¡£²»¹ý½¨ÒéÖÁÉÙÒªÌṩÁ½¸ö broker ÐÅÏ¢£¬Ò»µ©ÆäÖÐÒ»¸öå´»ú£¬Éú²úÕßÈÔÈ»Äܹ»Á¬½Óµ½¼¯ÈºÉÏ¡£
  • key.serializer
    • broker ÐèÒª½ÓÊÕµ½ÐòÁл¯Ö®ºóµÄ key/valueÖµ£¬365betÌåÓýÔÚÏßÉú²úÕß·¢Ë͵ÄÏûÏ¢ÐèÒª¾­¹ýÐòÁл¯Ö®ºó²Å´«µÝ¸ø Kafka Broker¡£Éú²úÕßÐèÒªÖªµÀ²ÉÓúÎÖÖ·½Ê½°Ñ Java ¶ÔÏóת»»Îª×Ö½ÚÊý×é¡£key.serializer ±ØÐë±»ÉèÖÃΪһ¸öʵÏÖÁËorg.apache.kafka.common.serialization.Serializer ½Ó¿ÚµÄÀ࣬Éú²úÕß»áʹÓÃÕâ¸öÀà°Ñ¼ü¶ÔÏóÐòÁл¯Îª×Ö½ÚÊý×é¡£ÕâÀïÍØչһϠSerializer Àà
    • Serializer ÊÇÒ»¸ö½Ó¿Ú£¬Ëü365ÌåÓýͶעÀཫ»á²ÉÓúÎÖÖ·½Ê½ÐòÁл¯£¬ËüµÄ×÷ÓÃÊǰѶÔÏóת»»Îª×Ö½Ú£¬ÊµÏÖÁË Serializer ½Ó¿ÚµÄÀàÖ÷ÒªÓÐ ByteArraySerializer¡¢StringSerializer¡¢IntegerSerializer £¬ÆäÖÐ ByteArraySerialize ÊÇ Kafka ĬÈÏʹÓõÄÐòÁл¯Æ÷£¬ÆäËûµÄÐòÁл¯Æ÷»¹Óкܶ࣬Äã¿ÉÒÔͨ¹ý ÕâÀï ²é¿´ÆäËûÐòÁл¯Æ÷¡£Òª×¢ÒâµÄÒ»µã£ºkey.serializer ÊDZØÐëÒªÉèÖõģ¬¼´Ê¹Äã´òËãÖ»·¢ËÍÖµµÄÄÚÈÝ¡£
  • value.serializer
    • Óë key.serializer Ò»Ñù£¬value.serializer Ö¸¶¨µÄÀà»á½«ÖµÐòÁл¯¡£
• ÏÂÃæ´úÂëÑÝʾÁËÈçºÎ´´½¨Ò»¸ö Kafka Éú²úÕߣ¬ÕâÀïÖ»Ö¸¶¨Á˱ØÒªµÄÊôÐÔ£¬ÆäËûʹÓÃĬÈϵÄÅäÖÃ

private Properties properties = new Properties();
properties.put("bootstrap.servers","broker1:9092,broker2:9092");
properties.put("key.serializer","org.apache.kafka.common.serialization.StringSerializer");
properties.put("value.serializer","org.apache.kafka.common.serialization.StringSerializer");
properties = new KafkaProducer<String,String>(properties);

• À´½âÊÍÒ»ÏÂÕâ¶Î´úÂë
  • Ê×ÏÈ´´½¨ÁËÒ»¸ö Properties ¶ÔÏó
  • ʹÓà StringSerializer ÐòÁл¯Æ÷ÐòÁл¯ key / value ¼üÖµ¶Ô
  • ÔÚÕâÀï365betÌåÓýÔÚÏßÃÇ´´½¨ÁËÒ»¸öеÄÉú²úÕ߶ÔÏ󣬲¢Îª¼üÖµÉèÖÃÁËÇ¡µ±µÄÀàÐÍ£¬È»ºó°Ñ Properties ¶ÔÏ󴫵ݸøËû¡£
• Kafka ÏûÏ¢·¢ËÍ
  • ʵÀý»¯Éú²úÕ߶ÔÏóºó£¬½ÓÏÂÀ´¾Í¿ÉÒÔ¿ªÊ¼·¢ËÍÏûÏ¢ÁË£¬·¢ËÍÏûÏ¢Ö÷ÒªÓÉÏÂÃ漸ÖÖ·½Ê½
• ¼òµ¥ÏûÏ¢·¢ËÍ
• Kafka ×î¼òµ¥µÄÏûÏ¢·¢ËÍÈçÏ£º

ProducerRecord<String,String> record =
¡¡¡¡¡¡¡¡¡¡¡¡¡¡¡¡¡¡new ProducerRecord<String, String>("CustomerCountry","West","France");

producer.send(record);

• ´úÂëÖÐÉú²úÕß(producer)µÄ send() ·½·¨ÐèÒª°Ñ ProducerRecord µÄ¶ÔÏó×÷Ϊ²ÎÊý½øÐз¢ËÍ£¬ProducerRecord Óкܶ๹Ô캯Êý£¬Õâ¸ö365betÌåÓýÔÚÏßÃÇÏÂÃæÌÖÂÛ£¬ÕâÀïµ÷ÓõÄÊÇ

public ProducerRecord(String topic, K key, V value) {}

• Õâ¸ö¹¹Ô캯Êý£¬ÐèÒª´«µÝµÄÊÇ topicÖ÷Ì⣬key ºÍ value¡£
• °Ñ¶ÔÓ¦µÄ²ÎÊý´«µÝÍê³Éºó£¬Éú²úÕßµ÷Óà send() ·½·¨·¢ËÍÏûÏ¢£¨ProducerRecord¶ÔÏ󣩡£365betÌåÓýÔÚÏßÃÇ¿ÉÒÔ´ÓÉú²úÕߵļܹ¹Í¼Öп´³ö£¬ÏûÏ¢ÊÇÏȱ»Ð´Èë·ÖÇøÖеĻº³åÇøÖУ¬È»ºó·ÖÅú´Î·¢Ë͸ø Kafka Broker¡£

• ·¢Ëͳɹ¦ºó£¬send() ·½·¨»á·µ»ØÒ»¸ö Future(java.util.concurrent) ¶ÔÏó£¬Future ¶ÔÏóµÄÀàÐÍÊÇ RecordMetadata ÀàÐÍ£¬365betÌåÓýÔÚÏßÃÇÉÏÃæÕâ¶Î´úÂëûÓп¼ÂÇ·µ»ØÖµ£¬365betÌåÓýÔÚÏßûÓÐÉú³É¶ÔÓ¦µÄ Future ¶ÔÏó£¬365betÌåÓýÔÚÏßûÓа취֪µÀÏûÏ¢ÊÇ·ñ·¢Ëͳɹ¦¡£Èç¹û²»ÊǺÜÖØÒªµÄÐÅÏ¢»òÕ߶Խá¹û²»»á²úÉúÓ°ÏìµÄÐÅÏ¢£¬¿ÉÒÔʹÓÃÕâÖÖ·½Ê½½øÐз¢ËÍ¡£
• 365betÌåÓýÔÚÏßÃÇ¿ÉÒÔºöÂÔ·¢ËÍÏûϢʱ¿ÉÄÜ·¢ÉúµÄ´íÎó»òÕßÔÚ·þÎñÆ÷¶Ë¿ÉÄÜ·¢ÉúµÄ´íÎ󣬵«ÔÚÏûÏ¢·¢ËÍ֮ǰ£¬Éú²úÕß»¹¿ÉÄÜ·¢ÉúÆäËûµÄÒì³£¡£ÕâЩÒì³£ÓпÉÄÜÊÇ SerializationException(ÐòÁл¯Ê§°Ü)£¬BufferedExhaustedException »ò TimeoutException(˵Ã÷»º³åÇøÒÑÂú)£¬ÓÖ»òÊÇ InterruptedException(˵Ã÷·¢ËÍÏ̱߳»ÖжÏ)
• ͬ²½·¢ÏûÏ¢
  • µÚ¶þÖÖÏûÏ¢·¢ËÍ»úÖÆÈçÏÂËùʾ

ProducerRecord<String,String> record =
                new ProducerRecord<String, String>("CustomerCountry","West","France");

try{
  RecordMetadata recordMetadata = producer.send(record).get();
}catch(Exception e){
  e.printStackTrace()£»
}

• ÕâÖÖ·¢ËÍÏûÏ¢µÄ·½Ê½½ÏÉÏÃæµÄ·¢ËÍ·½Ê½ÓÐÁ˸Ľø£¬Ê×Ïȵ÷Óà send() ·½·¨£¬È»ºóÔÙµ÷Óà get() ·½·¨µÈ´ý Kafka ÏìÓ¦¡£Èç¹û·þÎñÆ÷·µ»Ø´íÎó£¬get() ·½·¨»áÅ׳öÒì³££¬Èç¹ûûÓз¢Éú´íÎó£¬365betÌåÓýÔÚÏßÃÇ»áµÃµ½ RecordMetadata ¶ÔÏ󣬿ÉÒÔÓÃËüÀ´²é¿´ÏûÏ¢¼Ç¼¡£
• Éú²úÕߣ¨KafkaProducer£©ÔÚ·¢Ë͵Ĺý³ÌÖлá³öÏÖÁ½Àà´íÎó£ºÆäÖÐÒ»ÀàÊÇÖØÊÔ´íÎó£¬ÕâÀà´íÎó¿ÉÒÔͨ¹ýÖØ·¢ÏûÏ¢À´½â¾ö¡£±ÈÈçÁ¬½ÓµÄ´íÎ󣬿ÉÒÔͨ¹ýÔٴν¨Á¢Á¬½ÓÀ´½â¾ö£»ÎÞÖ÷´íÎóÔò¿ÉÒÔͨ¹ýÖØÐÂΪ·ÖÇøÑ¡¾ÙÊ×ÁìÀ´½â¾ö¡£KafkaProducer ±»ÅäÖÃΪ×Ô¶¯ÖØÊÔ£¬Èç¹û¶à´ÎÖØÊÔºóÈÔÎÞ·¨½â¾öÎÊÌ⣬Ôò»áÅ׳öÖØÊÔÒì³£¡£ÁíÒ»Àà´íÎóÊÇÎÞ·¨Í¨¹ýÖØÊÔÀ´½â¾öµÄ£¬±ÈÈçÏûÏ¢¹ý´ó¶ÔÓÚÕâÀà´íÎó£¬KafkaProducer ²»»á½øÐÐÖØÊÔ£¬Ö±½ÓÅ׳öÒì³£¡£
• Òì²½·¢ËÍÏûÏ¢
  • ͬ²½·¢ËÍÏûÏ¢¶¼ÓиöÎÊÌ⣬ÄǾÍÊÇͬһʱ¼äÖ»ÄÜÓÐÒ»¸öÏûÏ¢ÔÚ·¢ËÍ£¬Õâ»áÔì³ÉÐí¶àÏûÏ¢ÎÞ·¨Ö±½Ó·¢ËÍ£¬Ôì³ÉÏûÏ¢Öͺó£¬ÎÞ·¨·¢»ÓЧÒæ×î´ó»¯¡£
  • ±ÈÈçÏûÏ¢ÔÚÓ¦ÓóÌÐòºÍ Kafka ¼¯ÈºÖ®¼äÒ»¸öÀ´»ØÐèÒª 10ms¡£Èç¹û·¢ËÍÍê365ÌåÓýͶעÏûÏ¢ºó¶¼µÈ´ýÏìÓ¦µÄ»°£¬ÄÇô·¢ËÍ100¸öÏûÏ¢ÐèÒª 1 Ã룬µ«ÊÇÈç¹ûÊÇÒì²½·½Ê½µÄ»°£¬·¢ËÍ 100 ÌõÏûÏ¢ËùÐèÒªµÄʱ¼ä¾Í»áÉٺܶàºÜ¶à¡£´ó¶àÊýʱºò£¬ËäÈ»Kafka »á·µ»Ø RecordMetadata ÏûÏ¢£¬µ«ÊÇ365betÌåÓýÔÚÏßÃDz¢²»ÐèÒªµÈ´ýÏìÓ¦¡£
  • ΪÁËÔÚÒì²½·¢ËÍÏûÏ¢µÄͬʱÄܹ»¶ÔÒì³£Çé¿ö½øÐд¦Àí£¬Éú²úÕßÌṩÁ˻صôÖ§³Ö¡£ÏÂÃæÊǻص÷µÄÒ»¸öÀý×Ó

ProducerRecord<String, String> producerRecord = new ProducerRecord<String, String>("CustomerCountry", "Huston", "America");
        producer.send(producerRecord,new DemoProducerCallBack());


class DemoProducerCallBack implements Callback {

  public void onCompletion(RecordMetadata metadata, Exception exception) {
    if(exception != null){
      exception.printStackTrace();;
    }
  }
}

• Ê×ÏÈʵÏֻص÷ÐèÒª¶¨ÒåÒ»¸öʵÏÖÁËorg.apache.kafka.clients.producer.CallbackµÄÀ࣬Õâ¸ö½Ó¿ÚÖ»ÓÐÒ»¸ö onCompletion·½·¨¡£Èç¹û kafka ·µ»ØÒ»¸ö´íÎó£¬onCompletion ·½·¨»áÅ׳öÒ»¸ö·Ç¿Õ(non null)Òì³££¬ÕâÀï365betÌåÓýÔÚÏßÃÇÖ»ÊǼòµ¥µÄ°ÑËü´òÓ¡³öÀ´£¬Èç¹ûÊÇÉú²ú»·¾³ÐèÒª¸üÏêϸµÄ´¦Àí£¬È»ºóÔÚ send() ·½·¨·¢Ë͵Äʱºò´«µÝÒ»¸ö Callback »Øµ÷µÄ¶ÔÏó¡£
• Éú²úÕß·ÖÇø»úÖÆ
  • Kafka ¶ÔÓÚÊý¾ÝµÄ¶ÁдÊÇÒÔ·ÖÇøΪÁ£¶ÈµÄ£¬·ÖÇø¿ÉÒÔ·Ö²¼ÔÚ¶à¸öÖ÷»ú£¨Broker£©ÖУ¬ÕâÑù365ÌåÓýͶע½ÚµãÄܹ»ÊµÏÖ¶ÀÁ¢µÄÊý¾ÝдÈëºÍ¶ÁÈ¡£¬²¢ÇÒÄܹ»Í¨¹ýÔö¼ÓеĽڵãÀ´Ôö¼Ó Kafka ¼¯ÈºµÄÍÌÍÂÁ¿£¬Í¨¹ý·ÖÇø²¿ÊðÔÚ¶à¸ö Broker À´ÊµÏÖ¸ºÔؾùºâµÄЧ¹û¡£
  • ÉÏÃæ365betÌåÓýÔÚÏßÃǽéÉÜÁËÉú²úÕߵķ¢ËÍ·½Ê½ÓÐÈýÖÖ£º²»¹Ü½á¹ûÈçºÎÖ±½Ó·¢ËÍ¡¢·¢ËͲ¢·µ»Ø½á¹û¡¢·¢ËͲ¢»Øµ÷¡£ÓÉÓÚÏûÏ¢ÊÇ´æÔÚÖ÷Ì⣨topic£©µÄ·ÖÇø£¨partition£©Öеģ¬365betÌåÓýÔÚÏßµ± Producer Éú²úÕß·¢ËͲúÉúÒ»ÌõÏûÏ¢·¢¸ø topic µÄʱºò£¬ÄãÈçºÎÅжÏÕâÌõÏûÏ¢»á´æÔÚÄĸö·ÖÇøÖÐÄØ£¿
  • ÕâÆäʵ¾ÍÉè¼Æµ½ Kafka µÄ·ÖÇø»úÖÆÁË¡£
• ·ÖÇø²ßÂÔ
  • Kafka µÄ·ÖÇø²ßÂÔÖ¸µÄ¾ÍÊǽ«Éú²úÕß·¢Ë͵½Äĸö·ÖÇøµÄËã·¨¡£Kafka Ϊ365betÌåÓýÔÚÏßÃÇÌṩÁËĬÈϵķÖÇø²ßÂÔ£¬Í¬Ê±ËüÒ²Ö§³ÖÄã×Ô¶¨Òå·ÖÇø²ßÂÔ¡£
  • Èç¹ûÒª×Ô¶¨Òå·ÖÇø²ßÂԵĻ°£¬ÄãÐèÒªÏÔʾÅäÖÃÉú²úÕ߶˵IJÎÊý Partitioner.class£¬365betÌåÓýÔÚÏßÃÇ¿ÉÒÔ¿´Ò»ÏÂÕâ¸öÀàËüλÓÚ org.apache.kafka.clients.producer °üÏÂ

public interface Partitioner extends Configurable, Closeable {
  
  public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster);

  public void close();
  
  default public void onNewBatch(String topic, Cluster cluster, int prevPartition) {}
}

• Partitioner ÀàÓÐÈý¸ö·½·¨£¬·Ö±ðÀ´½âÊÍÒ»ÏÂ
  • partition(): Õâ¸öÀàÓм¸¸ö²ÎÊý: topic£¬365ÌåÓýͶעÐèÒª´«µÝµÄÖ÷Ì⣻key 365ÌåÓýͶעÏûÏ¢ÖеļüÖµ£»keyBytes365ÌåÓýͶע·ÖÇøÖÐÐòÁл¯¹ýºóµÄkey£¬byteÊý×éµÄÐÎʽ´«µÝ£»value 365ÌåÓýͶעÏûÏ¢µÄ value Öµ£»valueBytes 365ÌåÓýͶע·ÖÇøÖÐÐòÁл¯ºóµÄÖµÊý×飻cluster365ÌåÓýͶעµ±Ç°¼¯ÈºµÄÔ­Êý¾Ý¡£Kafka ¸øÄãÕâô¶àÐÅÏ¢£¬¾ÍÊÇÏ£ÍûÈÃÄãÄܹ»³ä·ÖµØÀûÓÃÕâЩÐÅÏ¢¶ÔÏûÏ¢½øÐзÖÇø£¬¼ÆËã³öËüÒª±»·¢Ë͵½Äĸö·ÖÇøÖС£
  • close() : ¼Ì³ÐÁË Closeable ½Ó¿ÚÄܹ»ÊµÏÖ close() ·½·¨£¬ÔÚ·ÖÇø¹Ø±Õʱµ÷Óá£
  • onNewBatch(): 365ÌåÓýͶע֪ͨ·ÖÇø³ÌÐòÓÃÀ´´´½¨ÐµÄÅú´Î
  • ÆäÖÐÓë·ÖÇø²ßÂÔϢϢÏà¹ØµÄ¾ÍÊÇ partition() ·½·¨ÁË£¬·ÖÇø²ßÂÔÓÐÏÂÃæÕ⼸ÖÖ
• ˳ÐòÂÖѯ
  • ˳Ðò·ÖÅ䣬ÏûÏ¢ÊǾùÔȵķÖÅä¸ø365ÌåÓýͶע partition£¬¼´365ÌåÓýͶע·ÖÇø´æ´¢Ò»´ÎÏûÏ¢¡£¾ÍÏñÏÂÃæÕâÑù

• • ÉÏͼ365ÌåÓýͶעµÄ¾ÍÊÇÂÖѯ²ßÂÔ£¬ÂÖѵ²ßÂÔÊÇ Kafka Producer ÌṩµÄĬÈϲßÂÔ£¬Èç¹ûÄ㲻ʹÓÃÖ¸¶¨µÄÂÖѵ²ßÂԵĻ°£¬Kafka ĬÈÏ»áʹÓÃ˳ÐòÂÖѵ²ßÂԵķ½Ê½¡£
• Ëæ»úÂÖѯ
  • Ëæ»úÂÖѯ¼ò¶øÑÔÖ®¾ÍÊÇËæ»úµÄÏò partition Öб£´æÏûÏ¢£¬ÈçÏÂͼËùʾ

• ʵÏÖËæ»ú·ÖÅäµÄ´úÂëÖ»ÐèÒªÁ½ÐУ¬ÈçÏÂ

List<PartitionInfo> partitions = cluster.partitionsForTopic(topic);
return ThreadLocalRandom.current().nextInt(partitions.size());

• ÏȼÆËã³ö¸ÃÖ÷Ìâ×ܵķÖÇøÊý£¬È»ºóËæ»úµØ·µ»ØÒ»¸öСÓÚËüµÄÕýÕûÊý¡£
• ±¾ÖÊÉÏ¿´Ëæ»ú²ßÂÔÒ²ÊÇÁ¦Çó½«Êý¾Ý¾ùÔȵشòÉ¢µ½¸÷¸ö·ÖÇø£¬µ«´Óʵ¼Ê±íÏÖÀ´¿´£¬ËüҪѷÓÚÂÖѯ²ßÂÔ£¬365betÌåÓýÔÚÏßÈç¹û×·ÇóÊý¾ÝµÄ¾ùÔÈ·Ö²¼£¬»¹ÊÇʹÓÃÂÖѯ²ßÂԱȽϺá£ÊÂʵÉÏ£¬Ëæ»ú²ßÂÔÊÇÀÏ°æ±¾Éú²úÕßʹÓõķÖÇø²ßÂÔ£¬ÔÚа汾ÖÐÒѾ­¸ÄΪÂÖѯÁË¡£
• °´ÕÕ key ½øÐÐÏûÏ¢±£´æ
  • Õâ¸ö²ßÂÔÒ²½Ð×ö key-ordering ²ßÂÔ£¬Kafka ÖÐÿÌõÏûÏ¢¶¼»áÓÐ×Ô¼ºµÄkey£¬Ò»µ©ÏûÏ¢±»¶¨ÒåÁË Key£¬ÄÇôÄã¾Í¿ÉÒÔ±£Ö¤Í¬Ò»¸ö Key µÄËùÓÐÏûÏ¢¶¼½øÈëµ½ÏàͬµÄ·ÖÇøÀïÃ棬ÓÉÓÚ365ÌåÓýͶע·ÖÇøϵÄÏûÏ¢´¦Àí¶¼ÊÇÓÐ˳ÐòµÄ£¬¹ÊÕâ¸ö²ßÂÔ±»³ÆΪ°´ÏûÏ¢¼ü±£Ðò²ßÂÔ£¬ÈçÏÂͼËùʾ

• ʵÏÖÕâ¸ö²ßÂ﵀ partition ·½·¨Í¬Ñù¼òµ¥£¬Ö»ÐèÒªÏÂÃæÁ½ÐдúÂë¼´¿É£º

List<PartitionInfo> partitions = cluster.partitionsForTopic(topic);
return Math.abs(key.hashCode()) % partitions.size();

• ÉÏÃæÕ⼸ÖÖ·ÖÇø²ßÂÔ¶¼ÊDZȽϻù´¡µÄ²ßÂÔ£¬³ý´ËÖ®Í⣬Ä㻹¿ÉÒÔ×Ô¶¨Òå·ÖÇø²ßÂÔ¡£
• Éú²úÕßѹËõ»úÖÆ
  • ѹËõÒ»´Ê¼òµ¥À´½²¾ÍÊÇ365ÌåÓýͶע»¥»»Ë¼Ï룬ËüÊÇ365ÌåÓýͶע¾­µäµÄÓà CPU ʱ¼äÈ¥»»´ÅÅÌ¿Õ¼ä»òÕß I/O ´«ÊäÁ¿µÄ˼Ï룬ϣÍûÒÔ½ÏСµÄ CPU ¿ªÏú´øÀ´¸üÉٵĴÅÅÌÕ¼Óûò¸üÉÙµÄÍøÂç I/O ´«Êä¡£Èç¹ûÄ㻹²»Á˽âµÄ»°365betÌåÓýÔÚÏßÏ£ÍûÄãÏȶÁÍêÕâƪÎÄÕ ³ÌÐòÔ±ÐèÒªÁ˽âµÄÓ²ºË֪ʶ֮ѹËõËã·¨£¬È»ºóÄã¾ÍÃ÷°×ѹËõÊÇÔõô»ØÊÂÁË¡£
• Kafka ѹËõÊÇʲô
  • Kafka µÄÏûÏ¢·ÖΪÁ½²ã£ºÏûÏ¢¼¯ºÏ ºÍ ÏûÏ¢¡£Ò»¸öÏûÏ¢¼¯ºÏÖаüº¬Èô¸ÉÌõÈÕÖ¾Ï¶øÈÕÖ¾Ïî²ÅÊÇÕæÕý·â×°ÏûÏ¢µÄµØ·½¡£Kafka µ×²ãµÄÏûÏ¢ÈÕÖ¾ÓÉһϵÁÐÏûÏ¢¼¯ºÏÈÕÖ¾Ïî×é³É¡£Kafka ͨ³£²»»áÖ±½Ó²Ù×÷¾ßÌåµÄÒ»ÌõÌõÏûÏ¢£¬Ëü×ÜÊÇÔÚÏûÏ¢¼¯ºÏÕâ¸ö²ãÃæÉϽøÐÐдÈë²Ù×÷¡£
  • ÔÚ Kafka ÖУ¬Ñ¹Ëõ»á·¢ÉúÔÚÁ½¸öµØ·½£ºKafka Producer ºÍ Kafka Consumer£¬ÎªÊ²Ã´ÆôÓÃѹËõ£¿Ëµ°×Á˾ÍÊÇÏûϢ̫´ó£¬ÐèÒª±äСһµã À´Ê¹ÏûÏ¢·¢µÄ¸ü¿ìһЩ¡£
  • Kafka Producer ÖÐʹÓà compression.type À´¿ªÆôѹËõ

private Properties properties = new Properties();
properties.put("bootstrap.servers","192.168.1.9:9092");
properties.put("key.serializer","org.apache.kafka.common.serialization.StringSerializer");
properties.put("value.serializer","org.apache.kafka.common.serialization.StringSerializer");
properties.put("compression.type", "gzip");

Producer<String,String> producer = new KafkaProducer<String, String>(properties);

ProducerRecord<String,String> record =
new ProducerRecord<String, String>("CustomerCountry","Precision Products","France");

• ÉÏÃæ´úÂë±íÃ÷¸Ã Producer µÄѹËõË㷨ʹÓõÄÊÇ GZIP
• ÓÐѹËõ±ØÓнâѹËõ£¬Producer ʹÓÃѹËõË㷨ѹËõÏûÏ¢ºó²¢·¢Ë͸ø·þÎñÆ÷ºó£¬ÓÉ Consumer Ïû·ÑÕß½øÐнâѹËõ£¬ÒòΪ²ÉÓõĺÎÖÖѹËõËã·¨ÊÇËæ×Å key¡¢value Ò»Æð·¢Ë͹ýÈ¥µÄ£¬365betÌåÓýÔÚÏßÏû·ÑÕßÖªµÀ²ÉÓúÎÖÖѹËõËã·¨¡£
• Kafka ÖØÒª²ÎÊýÅäÖÃ
  • 365betÌåÓýÔÚÏßÃÇÖ÷Òª½éÉÜÁËһϠkafka ¼¯Èº´î½¨µÄ²ÎÊý£¬±¾ÆªÎÄÕÂ365betÌåÓýÔÚÏßÃÇÀ´½éÉÜһϠKafka Éú²úÕßÖØÒªµÄÅäÖã¬Éú²úÕßÓкܶà¿ÉÅäÖõIJÎÊý¡£
  • ÔÚÎĵµÀﶼÓÐ˵Ã÷£¬365betÌåÓýÔÚÏßÃǽéÉܼ¸¸öÔÚÄÚ´æʹÓá¢ÐÔÄܺͿɿ¿ÐÔ·½Ãæ¶ÔÉú²úÕßÓ°Ïì±È½Ï´óµÄ²ÎÊý½øÐÐ˵Ã÷
• key.serializer
  • ÓÃÓÚ key ¼üµÄÐòÁл¯£¬ËüʵÏÖÁË org.apache.kafka.common.serialization.Serializer ½Ó¿Ú
• value.serializer
  • ÓÃÓÚ value ÖµµÄÐòÁл¯£¬ÊµÏÖÁË org.apache.kafka.common.serialization.Serializer ½Ó¿Ú
• acks
  • acks ²ÎÊýÖ¸¶¨ÁËÒªÓжàÉÙ¸ö·ÖÇø¸±±¾½ÓÊÕÏûÏ¢£¬Éú²úÕß²ÅÈÏΪÏûÏ¢ÊÇдÈë³É¹¦µÄ¡£´Ë²ÎÊý¶ÔÏûÏ¢¶ªÊ§µÄÓ°Ïì½Ï´ó
  • Èç¹û acks = 0£¬¾Í365ÌåÓýͶעÉú²úÕßÒ²²»ÖªµÀ×Ô¼º²úÉúµÄÏûÏ¢ÊÇ·ñ±»·þÎñÆ÷½ÓÊÕÁË£¬Ëü²ÅÖªµÀËüд³É¹¦ÁË¡£Èç¹û·¢Ë͵Ä;ÖвúÉúÁË´íÎó£¬Éú²úÕßÒ²²»ÖªµÀ£¬ËüÒ²±È½Ïã±ƣ¬ÒòΪûÓзµ»ØÈκÎÏûÏ¢¡£Õâ¾ÍÀàËÆÓÚ UDP µÄÔËÊä²ãЭÒ飬ֻ¹Ü·¢£¬·þÎñÆ÷½ÓÊܲ»½ÓÊÜËüÒ²²»¹ØÐÄ¡£
  • Èç¹û acks = 1£¬Ö»Òª¼¯ÈºµÄ Leader ½ÓÊÕµ½ÏûÏ¢£¬¾Í»á¸øÉú²úÕß·µ»ØÒ»ÌõÏûÏ¢£¬¸æËßËüдÈë³É¹¦¡£Èç¹û·¢ËÍ;ÖÐÔì³ÉÁËÍøÂçÒì³£»òÕß Leader »¹Ã»Ñ¡¾Ù³öÀ´µÈÆäËûÇé¿öµ¼ÖÂÏûϢдÈëʧ°Ü£¬Éú²úÕß»áÊܵ½´íÎóÏûÏ¢£¬ÕâʱºòÉú²úÕßÍùÍù»áÔÙ´ÎÖØ·¢Êý¾Ý¡£ÒòΪÏûÏ¢µÄ·¢ËÍÒ²·ÖΪ ͬ²½ ºÍ Òì²½£¬Kafka ΪÁ˱£Ö¤ÏûÏ¢µÄ¸ßЧ´«Êä»á¾ö¶¨ÊÇͬ²½·¢ËÍ»¹ÊÇÒì²½·¢ËÍ¡£Èç¹ûÈÿͻ§¶ËµÈ´ý·þÎñÆ÷µÄÏìÓ¦£¨Í¨¹ýµ÷Óà Future ÖÐµÄ get() ·½·¨£©£¬ÏÔÈ»»áÔö¼ÓÑÓ³Ù£¬Èç¹û¿Í»§¶ËʹÓûص÷£¬¾Í»á½â¾öÕâ¸öÎÊÌâ¡£
  • Èç¹û acks = all£¬ÕâÖÖÇé¿öÏÂÊÇÖ»Óе±ËùÓвÎÓ븴ÖƵĽڵ㶼ÊÕµ½ÏûϢʱ£¬Éú²úÕ߲Żá½ÓÊÕµ½Ò»¸öÀ´×Ô·þÎñÆ÷µÄÏûÏ¢¡£²»¹ý£¬ËüµÄÑÓ³Ù±È acks =1 ʱ¸ü¸ß£¬ÒòΪ365betÌåÓýÔÚÏßÃÇÒªµÈ´ý²»Ö»Ò»¸ö·þÎñÆ÷½Úµã½ÓÊÕÏûÏ¢¡£
• buffer.memory
  • ´Ë²ÎÊýÓÃÀ´ÉèÖÃÉú²úÕßÄڴ滺³åÇøµÄ´óС£¬Éú²úÕßÓÃËü»º³åÒª·¢Ë͵½·þÎñÆ÷µÄÏûÏ¢¡£Èç¹ûÓ¦ÓóÌÐò·¢ËÍÏûÏ¢µÄËٶȳ¬¹ý·¢Ë͵½·þÎñÆ÷µÄËٶȣ¬»áµ¼ÖÂÉú²úÕ߿ռ䲻×ã¡£Õâ¸öʱºò£¬send() ·½·¨µ÷ÓÃҪô±»×èÈû£¬ÒªÃ´Å׳öÒì³££¬¾ßÌåÈ¡¾öÓÚ block.on.buffer.null ²ÎÊýµÄÉèÖá£
• compression.type
  • ´Ë²ÎÊýÀ´365ÌåÓýͶעÉú²úÕßÆôÓúÎÖÖѹËõËã·¨£¬Ä¬ÈÏÇé¿öÏ£¬ÏûÏ¢·¢ËÍʱ²»»á±»Ñ¹Ëõ¡£¸Ã²ÎÊý¿ÉÒÔÉèÖÃΪ snappy¡¢gzip ºÍ lz4£¬ËüÖ¸¶¨ÁËÏûÏ¢·¢Ë͸ø broker ֮ǰʹÓÃÄÄ365ÌåÓýͶעѹËõËã·¨½øÐÐѹËõ¡£ÏÂÃæÊǸ÷ѹËõËã·¨µÄ¶Ô±È 

• retries
  • Éú²úÕß´Ó·þÎñÆ÷ÊÕµ½µÄ´íÎóÓпÉÄÜÊÇÁÙʱÐԵĴíÎ󣨱ÈÈç·ÖÇøÕÒ²»µ½Ê×Á죩£¬ÔÚÕâÖÖÇé¿öÏ£¬reteis ²ÎÊýµÄÖµ¾ö¶¨ÁËÉú²úÕß¿ÉÒÔÖØ·¢µÄÏûÏ¢´ÎÊý£¬Èç¹û´ïµ½Õâ¸ö´ÎÊý£¬Éú²úÕß»á·ÅÆúÖØÊÔ²¢·µ»Ø´íÎó¡£Ä¬ÈÏÇé¿öÏ£¬Éú²úÕßÔÚÿ´ÎÖØÊÔÖ®¼äµÈ´ý 100ms£¬Õâ¸öµÈ´ý²ÎÊý¿ÉÒÔͨ¹ý retry.backoff.ms ½øÐÐÐ޸ġ£
• batch.size
  • µ±Óжà¸öÏûÏ¢ÐèÒª±»·¢Ë͵½Í¬Ò»¸ö·ÖÇøʱ£¬Éú²úÕß»á°ÑËüÃÇ·ÅÔÚͬһ¸öÅú´ÎÀï¡£¸Ã²ÎÊýÖ¸¶¨ÁËÒ»¸öÅú´Î¿ÉÒÔʹÓõÄÄÚ´æ´óС£¬°´ÕÕ×Ö½ÚÊý¼ÆËã¡£µ±Åú´Î±»ÌîÂú£¬Åú´ÎÀïµÄËùÓÐÏûÏ¢»á±»·¢ËͳöÈ¥¡£²»¹ýÉú²úÕß¾®²»Ò»¶¨¶¼»áµÈµ½Åú´Î±»ÌîÂú²Å·¢ËÍ£¬È365betÌåÓýÔÚÏßâÌõÊýµÄÏûÏ¢¶¼¿ÉÄܱ»·¢ËÍ¡£
• client.id
  • ´Ë²ÎÊý¿ÉÒÔÊÇÈ365betÌåÓýÔÚÏßâµÄ×Ö·û´®£¬·þÎñÆ÷»áÓÃËüÀ´Ê¶±ðÏûÏ¢µÄÀ´Ô´£¬Ò»°ãÅäÖÃÔÚÈÕÖ¾Àï
• max.in.flight.requests.per.connection
  • ´Ë²ÎÊýÖ¸¶¨ÁËÉú²úÕßÔÚÊÕµ½·þÎñÆ÷ÏìӦ֮ǰ¿ÉÒÔ·¢ËͶàÉÙÏûÏ¢£¬ËüµÄÖµÔ½¸ß£¬¾Í»áÕ¼ÓÃÔ½¶àµÄÄڴ棬²»¹ýÒ²»áÌá¸ßÍÌÍÂÁ¿¡£°ÑËüÉèΪ1 ¿ÉÒÔ±£Ö¤ÏûÏ¢ÊÇ°´ÕÕ·¢Ë͵Ä˳ÐòдÈë·þÎñÆ÷¡£
• timeout.ms¡¢request.timeout.ms ºÍ metadata.fetch.timeout.ms
• request.timeout.ms Ö¸¶¨ÁËÉú²úÕßÔÚ·¢ËÍÊý¾ÝʱµÈ´ý·þÎñÆ÷·µ»ØµÄÏìӦʱ¼ä£¬metadata.fetch.timeout.ms Ö¸¶¨ÁËÉú²úÕßÔÚ»ñÈ¡ÔªÊý¾Ý£¨±ÈÈçÄ¿±ê·ÖÇøµÄÊ×ÁìÊÇË­£©Ê±µÈ´ý·þÎñÆ÷·µ»ØÏìÓ¦µÄʱ¼ä¡£Èç¹ûµÈ´ýʱ¼ä³¬Ê±£¬Éú²úÕßҪôÖØÊÔ·¢ËÍÊý¾Ý£¬ÒªÃ´·µ»ØÒ»¸ö´íÎó¡£timeout.ms Ö¸¶¨ÁË broker µÈ´ýͬ²½¸±±¾·µ»ØÏûϢȷÈϵÄʱ¼ä£¬Óë asks µÄÅäÖÃÏàÆ¥Åä----Èç¹ûÔÚÖ¸¶¨Ê±¼äÄÚûÓÐÊÕµ½Í¬²½¸±±¾µÄÈ·ÈÏ£¬ÄÇô broker ¾Í»á·µ»ØÒ»¸ö´íÎó¡£
• max.block.ms
  • ´Ë²ÎÊýÖ¸¶¨ÁËÔÚµ÷Óà send() ·½·¨»òʹÓà partitionFor() ·½·¨»ñÈ¡ÔªÊý¾ÝʱÉú²úÕßµÄ×èÈûʱ¼äµ±Éú²úÕߵķ¢ËÍ»º³åÇøÒѲ¶£¬»òÕßûÓпÉÓõÄÔªÊý¾Ýʱ£¬ÕâЩ·½·¨¾Í»á×èÈû¡£ÔÚ×èÈûʱ¼ä´ïµ½ max.block.ms ʱ£¬Éú²úÕß»áÅ׳ö³¬Ê±Òì³£¡£
• max.request.size
  • ¸Ã²ÎÊýÓÃÓÚ¿ØÖÆÉú²úÕß·¢Ë͵ÄÇëÇó´óС¡£Ëü¿ÉÒÔÖ¸ÄÜ·¢Ë͵ĵ¥¸öÏûÏ¢µÄ×î´óÖµ£¬Ò²¿ÉÒÔÖ¸µ¥¸öÇëÇóÀïËùÓÐÏûÏ¢µÄ×Ü´óС¡£
• receive.buffer.bytes ºÍ send.buffer.bytes
  • Kafka ÊÇ»ùÓÚ TCP ʵÏֵģ¬ÎªÁ˱£Ö¤¿É¿¿µÄÏûÏ¢´«Ê䣬ÕâÁ½¸ö²ÎÊý·Ö±ðÖ¸¶¨ÁË TCP Socket ½ÓÊպͷ¢ËÍÊý¾Ý°üµÄ»º³åÇøµÄ´óС¡£Èç¹ûËüÃDZ»ÉèÖÃΪ -1£¬¾ÍʹÓòÙ×÷ϵͳµÄĬÈÏÖµ¡£Èç¹ûÉú²úÕß»òÏû·ÑÕßÓë broker ´¦ÓÚ²»Í¬µÄÊý¾ÝÖÐÐÄ£¬ÄÇô¿ÉÒÔÊʵ±Ôö´óÕâЩֵ¡£

 

Kafka Consumer

• Ó¦ÓóÌÐòʹÓà KafkaConsumer ´Ó Kafka Öж©ÔÄÖ÷Ìâ²¢½ÓÊÕÀ´×ÔÕâЩÖ÷ÌâµÄÏûÏ¢£¬È»ºóÔÙ°ÑËûÃDZ£´æÆðÀ´¡£Ó¦ÓóÌÐòÊ×ÏÈÐèÒª´´½¨Ò»¸ö KafkaConsumer ¶ÔÏ󣬶©ÔÄÖ÷Ìâ²¢¿ªÊ¼½ÓÊÜÏûÏ¢£¬ÑéÖ¤ÏûÏ¢²¢±£´æ½á¹û¡£Ò»¶Îʱ¼äºó£¬Éú²úÕßÍùÖ÷ÌâдÈëµÄËٶȳ¬¹ýÁËÓ¦ÓóÌÐòÑéÖ¤Êý¾ÝµÄËٶȣ¬Õâʱºò¸ÃÈçºÎ´¦Àí£¿Èç¹ûֻʹÓõ¥¸öÏû·ÑÕߵĻ°£¬Ó¦ÓóÌÐò»á¸ú²»ÉÏÏûÏ¢Éú³ÉµÄËٶȣ¬¾ÍÏñ¶à¸öÉú²úÕßÏñÏàͬµÄÖ÷ÌâдÈëÏûÏ¢Ò»Ñù£¬Õâʱºò¾ÍÐèÒª¶à¸öÏû·ÑÕß¹²Í¬²ÎÓëÏû·ÑÖ÷ÌâÖеÄÏûÏ¢£¬¶ÔÏûÏ¢½øÐзÖÁ÷´¦Àí¡£
• Kafka Ïû·ÑÕß´ÓÊôÓÚÏû·ÑÕßȺ×é¡£Ò»¸öȺ×éÖеÄÏû·ÑÕ߶©ÔĵĶ¼ÊÇÏàͬµÄÖ÷Ì⣬365ÌåÓýͶעÏû·ÑÕß½ÓÊÕÖ÷ÌâÒ»²¿·Ö·ÖÇøµÄÏûÏ¢¡£ÏÂÃæÊÇÒ»¸ö Kafka ·ÖÇøÏû·ÑʾÒâͼ

• ÉÏͼÖеÄÖ÷Ìâ T1 ÓÐËĸö·ÖÇø£¬·Ö±ðÊÇ·ÖÇø0¡¢·ÖÇø1¡¢·ÖÇø2¡¢·ÖÇø3£¬365betÌåÓýÔÚÏßÃÇ´´½¨Ò»¸öÏû·ÑÕßȺ×é1£¬Ïû·ÑÕßȺ×éÖÐÖ»ÓÐÒ»¸öÏû·ÑÕߣ¬Ëü¶©ÔÄÖ÷ÌâT1£¬½ÓÊÕµ½ T1 ÖеÄÈ«²¿ÏûÏ¢¡£ÓÉÓÚÒ»¸öÏû·ÑÕß´¦ÀíËĸöÉú²úÕß·¢Ë͵½·ÖÇøµÄÏûÏ¢£¬Ñ¹Á¦ÓÐЩ´ó£¬ÐèÒª°ïÊÖÀ´°ïæ·Öµ£ÈÎÎñ£¬ÓÚÊǾÍÑݱäΪÏÂͼ

• ÕâÑùÒ»À´£¬Ïû·ÑÕßµÄÏû·ÑÄÜÁ¦¾Í´ó´óÌá¸ßÁË£¬µ«ÊÇÔÚijЩ»·¾³Ï±ÈÈçÓû§²úÉúÏûÏ¢Ìرð¶àµÄʱºò£¬Éú²úÕß²úÉúµÄÏûÏ¢ÈÔ¾ÉÈÃÏû·ÑÕß³Ô²»Ïû£¬ÄǾͼÌÐøÔö¼ÓÏû·ÑÕß¡£

• ÈçÉÏͼËùʾ£¬365ÌåÓýͶע·ÖÇøËù²úÉúµÄÏûÏ¢Äܹ»±»365ÌåÓýͶעÏû·ÑÕßȺ×éÖеÄÏû·ÑÕßÏû·Ñ£¬Èç¹ûÏòÏû·ÑÕßȺ×éÖÐÔö¼Ó¸ü¶àµÄÏû·ÑÕߣ¬ÄÇô¶àÓàµÄÏû·ÑÕß½«»áÏÐÖã¬ÈçÏÂͼËùʾ

• ÏòȺ×éÖÐÔö¼ÓÏû·ÑÕßÊǺáÏòÉìËõÏû·ÑÄÜÁ¦µÄÖ÷Òª·½Ê½¡£×ܶøÑÔÖ®£¬365betÌåÓýÔÚÏßÃÇ¿ÉÒÔͨ¹ýÔö¼ÓÏû·Ñ×éµÄÏû·ÑÕßÀ´½øÐÐˮƽÀ©Õ¹ÌáÉýÏû·ÑÄÜÁ¦¡£ÕâÒ²ÊÇΪʲô½¨Òé´´½¨Ö÷ÌâʱʹÓñȽ϶àµÄ·ÖÇøÊý£¬ÕâÑù¿ÉÒÔÔÚÏû·Ñ¸ºÔظߵÄÇé¿öÏÂÔö¼ÓÏû·ÑÕßÀ´ÌáÉýÐÔÄÜ¡£ÁíÍ⣬Ïû·ÑÕßµÄÊýÁ¿²»Ó¦¸Ã±È·ÖÇøÊý¶à£¬ÒòΪ¶à³öÀ´µÄÏû·ÑÕßÊÇ¿ÕÏеģ¬Ã»ÓÐÈκΰïÖú¡£
• Kafka Ò»¸öºÜÖØÒªµÄÌØÐÔ¾ÍÊÇ£¬Ö»ÐèдÈëÒ»´ÎÏûÏ¢£¬¿ÉÒÔÖ§³ÖÈ365betÌåÓýÔÚÏßâ¶àµÄÓ¦ÓöÁÈ¡Õâ¸öÏûÏ¢¡£»»¾ä»°Ëµ£¬365ÌåÓýͶעӦÓö¼¿ÉÒÔ¶Áµ½È«Á¿µÄÏûÏ¢¡£ÎªÁËʹµÃ365ÌåÓýͶעӦÓö¼ÄܶÁµ½È«Á¿ÏûÏ¢£¬Ó¦ÓÃÐèÒªÓв»Í¬µÄÏû·Ñ×é¡£¶ÔÓÚÉÏÃæµÄÀý×Ó£¬¼ÙÈç365betÌåÓýÔÚÏßÃÇÐÂÔöÁËÒ»¸öеÄÏû·Ñ×é G2£¬¶øÕâ¸öÏû·Ñ×éÓÐÁ½¸öÏû·ÑÕߣ¬ÄÇô¾ÍÑݱäΪÏÂͼÕâÑù

• ÔÚÕâ¸ö³¡¾°ÖУ¬Ïû·Ñ×é G1 ºÍÏû·Ñ×é G2 ¶¼ÄÜÊÕµ½ T1 Ö÷ÌâµÄÈ«Á¿ÏûÏ¢£¬ÔÚÂß¼­ÒâÒåÉÏÀ´ËµËüÃÇÊôÓÚ²»Í¬µÄÓ¦Óá£
• ×ܽáÆðÀ´¾ÍÊÇÈç¹ûÓ¦ÓÃÐèÒª¶ÁÈ¡È«Á¿ÏûÏ¢£¬ÄÇôÇëΪ¸ÃÓ¦ÓÃÉèÖÃÒ»¸öÏû·Ñ×飻Èç¹û¸ÃÓ¦ÓÃÏû·ÑÄÜÁ¦²»×㣬ÄÇô¿ÉÒÔ¿¼ÂÇÔÚÕâ¸öÏû·Ñ×éÀïÔö¼ÓÏû·ÑÕß¡£

 

Ïû·ÑÕß×éºÍ·ÖÇøÖØƽºâ

• Ïû·ÑÕß×éÊÇʲô

  • Ïû·ÑÕß×飨Consumer Group£©ÊÇÓÉÒ»¸ö»ò¶à¸öÏû·ÑÕßʵÀý£¨Consumer Instance£©×é³ÉµÄȺ×飬¾ßÓпÉÀ©Õ¹ÐԺͿÉÈÝ´íÐÔµÄ365ÌåÓýͶע»úÖÆ¡£Ïû·ÑÕß×éÄÚµÄÏû·ÑÕß¹²ÏíÒ»¸öÏû·ÑÕß×éID£¬Õâ¸öID Ò²½Ð×ö Group ID£¬×éÄÚµÄÏû·ÑÕß¹²Í¬¶ÔÒ»¸öÖ÷Ìâ½øÐж©ÔĺÍÏû·Ñ£¬Í¬Ò»¸ö×éÖеÄÏû·ÑÕßÖ»ÄÜÏû·ÑÒ»¸ö·ÖÇøµÄÏûÏ¢£¬¶àÓàµÄÏû·ÑÕß»áÏÐÖã¬Åɲ»ÉÏÓó¡¡£
  • 365betÌåÓýÔÚÏßÃÇÔÚÉÏÃæÌáµ½ÁËÁ½ÖÖÏû·Ñ·½Ê½
  • Ò»¸öÏû·ÑÕßȺ×éÏû·ÑÒ»¸öÖ÷ÌâÖеÄÏûÏ¢£¬ÕâÖÖÏû·ÑģʽÓÖ³ÆΪµã¶ÔµãµÄÏû·Ñ·½Ê½£¬µã¶ÔµãµÄÏû·Ñ·½Ê½ÓÖ±»³ÆΪÏûÏ¢¶ÓÁÐ
  • Ò»¸öÖ÷ÌâÖеÄÏûÏ¢±»¶à¸öÏû·ÑÕßȺ×鹲ͬÏû·Ñ£¬ÕâÖÖÏû·ÑģʽÓÖ³ÆΪ·¢²¼-¶©ÔÄģʽ
• Ïû·ÑÕßÖØƽºâ
  • 365betÌåÓýÔÚÏßÃÇ´ÓÉÏÃæµÄÏû·ÑÕßÑݱäͼÖпÉÒÔÖªµÀÕâôһ¸ö¹ý³Ì£º×î³õÊÇÒ»¸öÏû·ÑÕ߶©ÔÄÒ»¸öÖ÷Ìâ²¢Ïû·ÑÆäÈ«²¿·ÖÇøµÄÏûÏ¢£¬ºóÀ´ÓÐÒ»¸öÏû·ÑÕß¼ÓÈëȺ×飬ËæºóÓÖÓиü¶àµÄÏû·ÑÕß¼ÓÈëȺ×飬¶øмÓÈëµÄÏû·ÑÕßʵÀý·Ö̯ÁË×î³õÏû·ÑÕߵIJ¿·ÖÏûÏ¢£¬ÕâÖÖ°Ñ·ÖÇøµÄËùÓÐȨͨ¹ýÒ»¸öÏû·ÑÕßתµ½ÆäËûÏû·ÑÕßµÄÐÐΪ³ÆΪÖØƽºâ£¬Ó¢ÎÄÃûÒ²½Ð×ö Rebalance ¡£ÈçÏÂͼËùʾ

• ÖØƽºâ·Ç³£ÖØÒª£¬ËüΪÏû·ÑÕßȺ×é´øÀ´ÁË¸ß¿ÉÓÃÐÔ ºÍ ÉìËõÐÔ£¬365betÌåÓýÔÚÏßÃÇ¿ÉÒÔ·ÅÐĵÄÌí¼ÓÏû·ÑÕß»òÒƳýÏû·ÑÕߣ¬²»¹ýÔÚÕý³£Çé¿öÏÂ365betÌåÓýÔÚÏßÃDz¢²»Ï£Íû·¢ÉúÕâÑùµÄÐÐΪ¡£ÔÚÖØƽºâÆڼ䣬Ïû·ÑÕßÎÞ·¨¶ÁÈ¡ÏûÏ¢£¬Ôì³ÉÕû¸öÏû·ÑÕß×éÔÚÖØƽºâµÄÆڼ䶼²»¿ÉÓá£ÁíÍ⣬µ±·ÖÇø±»ÖØзÖÅä¸øÁíÒ»¸öÏû·ÑÕßʱ£¬ÏûÏ¢µ±Ç°µÄ¶Áȡ״̬»á¶ªÊ§£¬ËüÓпÉÄÜ»¹ÐèҪȥˢлº´æ£¬ÔÚËüÖØлָ´×´Ì¬Ö®Ç°»áÍÏÂýÓ¦ÓóÌÐò¡£
• Ïû·ÑÕßͨ¹ýÏò×é֯Эµ÷Õß£¨Kafka Broker£©·¢ËÍÐÄÌøÀ´Î¬»¤×Ô¼ºÊÇÏû·ÑÕß×éµÄÒ»Ô±²¢È·ÈÏÆäÓµÓеķÖÇø¡£¶ÔÓÚ²»Í¬²»µÄÏû·ÑȺÌåÀ´Ëµ£¬Æä×é֯Эµ÷Õß¿ÉÒÔÊDz»Í¬µÄ¡£Ö»ÒªÏû·ÑÕ߶¨ÆÚ·¢ËÍÐÄÌø£¬¾Í»áÈÏΪÏû·ÑÕßÊÇ´æ»îµÄ²¢´¦ÀíÆä·ÖÇøÖеÄÏûÏ¢¡£µ±Ïû·ÑÕß¼ìË÷¼Ç¼»òÕßÌá½»ËüËùÏû·ÑµÄ¼Ç¼ʱ¾Í»á·¢ËÍÐÄÌø¡£
• Èç¹û¹ýÁËÒ»¶Îʱ¼ä Kafka Í£Ö¹·¢ËÍÐÄÌøÁË£¬»á»°£¨Session£©¾Í»á¹ýÆÚ£¬×é֯Эµ÷Õ߾ͻáÈÏΪÕâ¸ö Consumer ÒѾ­ËÀÍö£¬¾Í»á´¥·¢Ò»´ÎÖØƽºâ¡£Èç¹ûÏû·ÑÕßå´»ú²¢ÇÒÍ£Ö¹·¢ËÍÏûÏ¢£¬×é֯Эµ÷Õß»áµÈ´ý¼¸ÃëÖÓ£¬È·ÈÏËüËÀÍöÁ˲Żᴥ·¢ÖØƽºâ¡£ÔÚÕâ¶Îʱ¼äÀËÀÍöµÄÏû·ÑÕß½«²»´¦ÀíÈκÎÏûÏ¢¡£ÔÚÇåÀíÏû·ÑÕßʱ£¬Ïû·ÑÕß½«Í¨ÖªÐ­µ÷ÕßËüÒªÀ뿪Ⱥ×飬×é֯Эµ÷Õ߻ᴥ·¢Ò»´ÎÖØƽºâ£¬¾¡Á¿½µµÍ´¦ÀíÍ£¶Ù¡£
• ÖØƽºâÊÇÒ»°ÑË«Èн££¬ËüΪÏû·ÑÕßȺ×é´øÀ´¸ß¿ÉÓÃÐÔºÍÉìËõÐÔµÄͬʱ£¬»¹ÓÐÓÐһЩÃ÷ÏÔµÄȱµã(bug)£¬¶øÕâЩ bug µ½ÏÖÔÚÉçÇø»¹ÎÞ·¨Ð޸ġ£
• ÖØƽºâµÄ¹ý³Ì¶ÔÏû·ÑÕß×éÓм«´óµÄÓ°Ïì¡£ÒòΪÿ´ÎÖØƽºâ¹ý³ÌÖж¼»áµ¼ÖÂÍòÎï¾²Ö¹£¬²Î¿¼ JVM ÖеÄÀ¬»ø»ØÊÕ»úÖÆ£¬Ò²¾ÍÊÇ Stop The World £¬STW£¬(ÒýÓÃ×Ô¡¶ÉîÈëÀí½â Java ÐéÄâ»ú¡·ÖÐ p76 ¹ØÓÚ Serial ÊÕ¼¯Æ÷µÄÃèÊö)£º
• ¸üÖØÒªµÄÊÇËüÔÚ½øÐÐÀ¬»øÊÕ¼¯Ê±£¬±ØÐëÔÝÍ£ÆäËûËùÓеŤ×÷Ï̡߳£Ö±µ½ËüÊÕ¼¯½áÊø¡£Stop The World Õâ¸öÃû×ÖÌýÆðÀ´ºÜ˧£¬µ«ÕâÏ×÷ʵ¼ÊÉÏÊÇÓÉÐéÄâ»úÔÚºǫ́×Ô¶¯·¢Æð²¢Íê³ÉµÄ£¬ÔÚÓû§²»¿É¼ûµÄÇé¿öÏ°ÑÓû§Õý³£¹¤×÷µÄÏß³ÌÈ«²¿Í£µô£¬Õâ¶ÔºÜ¶àÓ¦ÓÃÀ´Ëµ¶¼ÊÇÄÑÒÔ½ÓÊܵġ£
• Ò²¾ÍÊÇ˵£¬ÔÚÖØƽºâÆڼ䣬Ïû·ÑÕß×éÖеÄÏû·ÑÕßʵÀý¶¼»áÍ£Ö¹Ïû·Ñ£¬µÈ´ýÖØƽºâµÄÍê³É¡£¶øÇÒÖØƽºâÕâ¸ö¹ý³ÌºÜÂý¡­
• ´´½¨Ïû·ÑÕß
  • ÉÏÃæµÄÀíÂÛ˵µÄÓеã¶à£¬ÏÂÃæ¾Íͨ¹ý´úÂëÀ´½²½âÒ»ÏÂÏû·ÑÕßÊÇÈçºÎÏû·ÑµÄ
  • ÔÚ¶ÁÈ¡ÏûϢ֮ǰ£¬ÐèÒªÏÈ´´½¨Ò»¸ö KafkaConsumer ¶ÔÏó¡£´´½¨ KafkaConsumer ¶ÔÏóÓë´´½¨ KafkaProducer ¶ÔÏóÊ®·ÖÏàËÆ ¡ª °ÑÐèÒª´«µÝ¸øÏû·ÑÕßµÄÊôÐÔ·ÅÔÚ properties ¶ÔÏóÖУ¬ºóÃæ365betÌåÓýÔÚÏßÃÇ»á×ÅÖØÌÖÂÛ Kafka µÄһЩÅäÖã¬ÕâÀï365betÌåÓýÔÚÏßÃÇÏȼòµ¥µÄ´´½¨Ò»Ï£¬Ê¹ÓÃ3¸öÊôÐÔ¾Í×ãÒÓ£¬·Ö±ðÊÇ bootstrap.server£¬key.deserializer£¬value.deserializer ¡£
  • »¹ÓÐÒ»¸öÊôÐÔÊÇ group.id Õâ¸öÊôÐÔ²»ÊDZØÐëµÄ£¬ËüÖ¸¶¨ÁË KafkaConsumer ÊÇÊôÓÚÄĸöÏû·ÑÕßȺ×é¡£´´½¨²»ÊôÓÚÈκ365betÌåÓýÔÚÏß»¸öȺ×éµÄÏû·ÑÕßÒ²ÊÇ¿ÉÒÔµÄ

Properties properties = new Properties();
        properties.put("bootstrap.server","192.168.1.9:9092");     properties.put("key.serializer","org.apache.kafka.common.serialization.StringSerializer");   properties.put("value.serializer","org.apache.kafka.common.serialization.StringSerializer");
KafkaConsumer<String,String> consumer = new KafkaConsumer<>(properties);

• Ö÷ÌⶩÔÄ

  • ´´½¨ºÃÏû·ÑÕßÖ®ºó£¬ÏÂÒ»²½¾Í¿ªÊ¼¶©ÔÄÖ÷ÌâÁË¡£subscribe() ·½·¨½ÓÊÜÒ»¸öÖ÷ÌâÁбí×÷Ϊ²ÎÊý£¬Ê¹ÓÃÆðÀ´±È½Ï¼òµ¥

consumer.subscribe(Collections.singletonList("customerTopic"));

• • ΪÁ˼òµ¥365betÌåÓýÔÚÏßÃÇÖ»¶©ÔÄÁËÒ»¸öÖ÷Ìâ customerTopic£¬²ÎÊý´«ÈëµÄÊÇÒ»¸öÕýÔò±í´ïʽ£¬ÕýÔò±í´ïʽ¿ÉÒÔÆ¥Åä¶à¸öÖ÷Ì⣬Èç¹ûÓÐÈË´´½¨ÁËеÄÖ÷Ì⣬²¢ÇÒÖ÷ÌâµÄÃû×ÖÓëÕýÔò±í´ïʽÏàÆ¥Å䣬ÄÇô»áÁ¢¼´´¥·¢Ò»´ÎÖØƽºâ£¬Ïû·ÑÕ߾ͿÉÒÔ¶ÁȡеÄÖ÷Ìâ¡£
  • Òª¶©ÔÄËùÓÐÓë test Ïà¹ØµÄÖ÷Ì⣬¿ÉÒÔÕâÑù×ö

consumer.subscribe("test.*");

• ÂÖѯ
  • 365betÌåÓýÔÚÏßÃÇÖªµÀ£¬Kafka ÊÇÖ§³Ö¶©ÔÄ/·¢²¼Ä£Ê½µÄ£¬Éú²úÕß·¢ËÍÊý¾Ý¸ø Kafka Broker£¬ÄÇôÏû·ÑÕßÊÇÈçºÎÖªµÀÉú²úÕß·¢ËÍÁËÊý¾ÝÄØ£¿ÆäʵÉú²úÕß²úÉúµÄÊý¾ÝÏû·ÑÕßÊDz»ÖªµÀµÄ£¬KafkaConsumer ²ÉÓÃÂÖѯµÄ·½Ê½¶¨ÆÚÈ¥ Kafka Broker ÖнøÐÐÊý¾ÝµÄ¼ìË÷£¬Èç¹ûÓÐÊý¾Ý¾ÍÓÃÀ´Ïû·Ñ£¬Èç¹ûûÓоÍÔÙ¼ÌÐøÂÖѯµÈ´ý£¬ÏÂÃæÊÇÂÖѯµÈ´ýµÄ¾ßÌåʵÏÖ

try {
  while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(100));
    for (ConsumerRecord<String, String> record : records) {
      int updateCount = 1;
      if (map.containsKey(record.value())) {
        updateCount = (int) map.get(record.value() + 1);
      }
      map.put(record.value(), updateCount);
    }
  }
}finally {
  consumer.close();
}

• ÕâÊÇÒ»¸öÎÞÏÞÑ­»·¡£Ïû·ÑÕßʵ¼ÊÉÏÊÇÒ»¸ö³¤ÆÚÔËÐеÄÓ¦ÓóÌÐò£¬Ëüͨ¹ýÂÖѯµÄ·½Ê½Ïò Kafka ÇëÇóÊý¾Ý¡£
• µÚÈýÐдúÂë·Ç³£ÖØÒª£¬Kafka ±ØÐ붨ÆÚÑ­»·ÇëÇóÊý¾Ý£¬·ñÔò¾Í»áÈÏΪ¸Ã Consumer ÒѾ­¹ÒÁË£¬»á´¥·¢ÖØƽºâ£¬ËüµÄ·ÖÇø»áÒƽ»¸øȺ×éÖеÄÆäËüÏû·ÑÕß¡£´«¸ø poll() ·½·¨µÄÊÇÒ»¸ö³¬ÊÐʱ¼ä£¬Óàjava.time.Duration ÀàÀ´365ÌåÓýͶע£¬Èç¹û¸Ã²ÎÊý±»ÉèÖÃΪ 0 £¬poll() ·½·¨»áÁ¢¿Ì·µ»Ø£¬·ñÔò¾Í»áÔÚÖ¸¶¨µÄºÁÃëÊýÄÚÒ»Ö±µÈ´ý broker ·µ»ØÊý¾Ý¡£
• poll() ·½·¨»á·µ»ØÒ»¸ö¼Ç¼ÁÐ±í¡£Ã¿Ìõ¼Ç¼¶¼°üº¬Á˼ǼËùÊôÖ÷ÌâµÄÐÅÏ¢£¬¼Ç¼ËùÔÚ·ÖÇøµÄÐÅÏ¢¡¢¼Ç¼ÔÚ·ÖÇøÖеÄÆ«ÒÆÁ¿£¬ÒÔ¼°¼Ç¼µÄ¼üÖµ¶Ô¡£365betÌåÓýÔÚÏßÃÇÒ»°ã»á±éÀúÕâ¸öÁÐ±í£¬ÖðÌõ´¦ÀíÿÌõ¼Ç¼¡£
• ÔÚÍ˳öÓ¦ÓóÌÐò֮ǰʹÓàclose() ·½·¨¹Ø±ÕÏû·ÑÕß¡£ÍøÂçÁ¬½ÓºÍ socket Ò²»áËæÖ®¹Ø±Õ£¬²¢Á¢¼´´¥·¢Ò»´ÎÖØƽºâ£¬¶ø²»ÊǵȴýȺ×éЭµ÷Æ÷·¢ÏÖËü²»ÔÙ·¢ËÍÐÄÌø²¢È϶¨ËüÒѾ­ËÀÍö¡£
• Ḭ̈߳²È«ÐÔ
  • ÔÚͬһ¸öȺ×éÖУ¬365betÌåÓýÔÚÏßÃÇÎÞ·¨ÈÃÒ»¸öÏß³ÌÔËÐжà¸öÏû·ÑÕߣ¬Ò²ÎÞ·¨Èöà¸öḬ̈߳²È«µÄ¹²ÏíÒ»¸öÏû·ÑÕß¡£°´ÕÕ¹æÔò£¬Ò»¸öÏû·ÑÕßʹÓÃÒ»¸öỊ̈߳¬Èç¹ûÒ»¸öÏû·ÑÕßȺ×éÖжà¸öÏû·ÑÕ߶¼ÏëÒªÔËÐеĻ°£¬ÄÇô±ØÐëÈÃ365ÌåÓýͶעÏû·ÑÕßÔÚ×Ô¼ºµÄÏß³ÌÖÐÔËÐУ¬¿ÉÒÔʹÓà Java ÖеĠExecutorService Æô¶¯¶à¸öÏû·ÑÕß½øÐнøÐд¦Àí¡£
• Ïû·ÑÕßÅäÖÃ
  • µ½Ä¿Ç°ÎªÖ¹£¬365betÌåÓýÔÚÏßÃÇѧϰÁËÈçºÎʹÓÃÏû·ÑÕß API£¬²»¹ýÖ»½éÉÜÁ˼¸¸ö×î»ù±¾µÄÊôÐÔ£¬Kafka ÎĵµÁгöÁËËùÓÐÓëÏû·ÑÕßÏà¹ØµÄÅäÖÃ˵Ã÷¡£´ó²¿·Ö²ÎÊý¶¼ÓкÏÀíµÄĬÈÏÖµ£¬Ò»°ã²»ÐèÒªÐÞ¸ÄËüÃÇ£¬ÏÂÃæ365betÌåÓýÔÚÏßÃǾÍÀ´½éÉÜÒ»ÏÂÕâЩ²ÎÊý¡£
• fetch.min.bytes
  • ¸ÃÊôÐÔÖ¸¶¨ÁËÏû·ÑÕß´Ó·þÎñÆ÷»ñÈ¡¼Ç¼µÄ×îС×Ö½ÚÊý¡£broker ÔÚÊÕµ½Ïû·ÑÕßµÄÊý¾ÝÇëÇóʱ£¬Èç¹û¿ÉÓõÄÊý¾ÝÁ¿Ð¡ÓÚ fetch.min.bytes Ö¸¶¨µÄ´óС£¬ÄÇôËü»áµÈµ½ÓÐ×ã¹»µÄ¿ÉÓÃÊý¾Ýʱ²Å°ÑËü·µ»Ø¸øÏû·ÑÕß¡£ÕâÑù¿ÉÒÔ½µµÍÏû·ÑÕßºÍ broker µÄ¹¤×÷¸ºÔØ£¬ÒòΪËüÃÇÔÚÖ÷ÌâʹÓÃƵÂʲ»ÊǺܸߵÄʱºò¾Í²»ÓÃÀ´»Ø´¦ÀíÏûÏ¢¡£Èç¹ûûÓкܶà¿ÉÓÃÊý¾Ý£¬µ«Ïû·ÑÕßµÄ CPU ʹÓÃÂʺܸߣ¬ÄÇô¾ÍÐèÒª°Ñ¸ÃÊôÐÔµÄÖµÉèµÃ±ÈĬÈÏÖµ´ó¡£Èç¹ûÏû·ÑÕßµÄÊýÁ¿±È½Ï¶à£¬°Ñ¸ÃÊôÐÔµÄÖµµ÷´ó¿ÉÒÔ½µµÍ broker µÄ¹¤×÷¸ºÔØ¡£
• fetch.max.wait.ms
  • 365betÌåÓýÔÚÏßÃÇͨ¹ýÉÏÃæµÄ fetch.min.bytes ¸æËß Kafka£¬µÈµ½ÓÐ×ã¹»µÄÊý¾Ýʱ²Å»á°ÑËü·µ»Ø¸øÏû·ÑÕß¡£¶ø fetch.max.wait.ms ÔòÓÃÓÚÖ¸¶¨ broker µÄµÈ´ýʱ¼ä£¬Ä¬ÈÏÊÇ 500 ºÁÃë¡£Èç¹ûûÓÐ×ã¹»µÄÊý¾ÝÁ÷Èë kafka µÄ»°£¬Ïû·ÑÕß»ñÈ¡µÄ×îСÊý¾ÝÁ¿ÒªÇó¾ÍµÃ²»µ½Âú×㣬×îÖÕµ¼Ö 500 ºÁÃëµÄÑÓ³Ù¡£Èç¹ûÒª½µµÍDZÔÚµÄÑÓ³Ù£¬¾Í¿ÉÒ԰ѲÎÊýÖµÉèÖõÄСһЩ¡£Èç¹û fetch.max.wait.ms ±»ÉèÖÃΪ 100 ºÁÃëµÄÑÓ³Ù£¬¶ø fetch.min.bytes µÄÖµÉèÖÃΪ 1MB£¬ÄÇô Kafka ÔÚÊÕµ½Ïû·ÑÕßÇëÇóºó£¬ÒªÃ´·µ»Ø 1MB µÄÊý¾Ý£¬ÒªÃ´ÔÚ 100 ms ºó·µ»ØËùÓпÉÓõÄÊý¾Ý¡£¾Í¿´ÄĸöÌõ¼þÊ×Ïȱ»Âú×ã¡£
• max.partition.fetch.bytes
  • ¸ÃÊôÐÔÖ¸¶¨ÁË·þÎñÆ÷´Ó365ÌåÓýͶע·ÖÇøÀï·µ»Ø¸øÏû·ÑÕßµÄ×î´ó×Ö½ÚÊý¡£ËüµÄĬÈÏֵʱ 1MB£¬Ò²¾ÍÊÇ˵£¬KafkaConsumer.poll() ·½·¨´Ó365ÌåÓýͶע·ÖÇøÀï·µ»ØµÄ¼Ç¼×î¶à²»³¬¹ý max.partition.fetch.bytes Ö¸¶¨µÄ×Ö½Ú¡£Èç¹ûÒ»¸öÖ÷ÌâÓÐ20¸ö·ÖÇøºÍ5¸öÏû·ÑÕߣ¬ÄÇô365ÌåÓýͶעÏû·ÑÕßÐèÒªÖÁÉÙ4 MBµÄ¿ÉÓÃÄÚ´æÀ´½ÓÊռǼ¡£ÔÚΪÏû·ÑÕß·ÖÅäÄÚ´æʱ£¬¿ÉÒÔ¸øËüÃǶà·ÖÅäһЩ£¬ÒòΪÈç¹ûȺ×éÀïÓÐÏû·ÑÕß·¢Éú±ÀÀ££¬Ê£ÏµÄÏû·ÑÕßÐèÒª´¦Àí¸ü¶àµÄ·ÖÇø¡£max.partition.fetch.bytes µÄÖµ±ØÐë±È broker Äܹ»½ÓÊÕµÄ×î´óÏûÏ¢µÄ×Ö½ÚÊý(ͨ¹ý max.message.size ÊôÐÔÅäÖôó)£¬·ñÔòÏû·ÑÕß¿ÉÄÜÎÞ·¨¶ÁÈ¡ÕâЩÏûÏ¢£¬µ¼ÖÂÏû·ÑÕßÒ»Ö±¹ÒÆðÖØÊÔ¡£ ÔÚÉèÖøÃÊôÐÔʱ£¬ÁíÍâÒ»¸ö¿¼Á¿µÄÒòËØÊÇÏû·ÑÕß´¦ÀíÊý¾ÝµÄʱ¼ä¡£Ïû·ÑÕßÐèҪƵ·±µÄµ÷Óà poll() ·½·¨À´±ÜÃâ»á»°¹ýÆںͷ¢Éú·ÖÇøÔÙƽºâ£¬Èç¹ûµ¥´Îµ÷ÓÃpoll() ·µ»ØµÄÊý¾ÝÌ«¶à£¬Ïû·ÑÕßÐèÒª¸ü¶àµÄʱ¼ä½øÐд¦Àí£¬¿ÉÄÜÎÞ·¨¼°Ê±½øÐÐÏÂÒ»¸öÂÖѯÀ´±ÜÃâ»á»°¹ýÆÚ¡£Èç¹û³öÏÖÕâÖÖÇé¿ö£¬¿ÉÒÔ°Ñ max.partition.fetch.bytes Öµ¸ÄС£¬»òÕßÑÓ³¤»á»°¹ýÆÚʱ¼ä¡£
• session.timeout.ms
  • Õâ¸öÊôÐÔÖ¸¶¨ÁËÏû·ÑÕßÔÚ±»ÈÏΪËÀÍö֮ǰ¿ÉÒÔÓë·þÎñÆ÷¶Ï¿ªÁ¬½ÓµÄʱ¼ä£¬Ä¬ÈÏÊÇ 3s¡£Èç¹ûÏû·ÑÕßûÓÐÔÚ session.timeout.ms Ö¸¶¨µÄʱ¼äÄÚ·¢ËÍÐÄÌø¸øȺ×éЭµ÷Æ÷£¬¾Í»á±»È϶¨ÎªËÀÍö£¬Ð­µ÷Æ÷¾Í»á´¥·¢ÖØƽºâ¡£°ÑËüµÄ·ÖÇø·ÖÅä¸øÏû·ÑÕßȺ×éÖеÄÆäËüÏû·ÑÕߣ¬´ËÊôÐÔÓë heartbeat.interval.ms ½ôÃÜÏà¹Ø¡£heartbeat.interval.ms Ö¸¶¨ÁË poll() ·½·¨ÏòȺ×éЭµ÷Æ÷·¢ËÍÐÄÌøµÄƵÂÊ£¬session.timeout.ms ÔòÖ¸¶¨ÁËÏû·ÑÕß¿ÉÒÔ¶à¾Ã²»·¢ËÍÐÄÌø¡£365betÌåÓýÔÚÏߣ¬ÕâÁ½¸öÊôÐÔÒ»°ãÐèҪͬʱÐ޸ģ¬heartbeat.interval.ms ±ØÐë±È session.timeout.ms С£¬Ò»°ãÊÇ session.timeout.ms µÄÈý·ÖÖ®Ò»¡£Èç¹û session.timeout.ms ÊÇ 3s£¬ÄÇô heartbeat.interval.ms Ó¦¸ÃÊÇ 1s¡£°Ñ session.timeout.ms ÖµÉèÖõıÈĬÈÏֵС£¬¿ÉÒÔ¸ü¿ìµØ¼ì²âºÍ»Ö¸´±À·ßµÄ½Úµã£¬²»¹ý³¤Ê±¼äµÄÂÖѯ»òÀ¬»øÊÕ¼¯¿ÉÄܵ¼Ö·ÇÔ¤ÆÚµÄÖØƽºâ¡£°Ñ¸ÃÊôÐÔµÄÖµÉèÖõôóһЩ£¬¿ÉÒÔ¼õÉÙÒâÍâµÄÖØƽºâ£¬²»¹ý¼ì²â½Úµã±ÀÀ£ÐèÒª¸ü³¤µÄʱ¼ä¡£
• auto.offset.reset
  • ¸ÃÊôÐÔÖ¸¶¨ÁËÏû·ÑÕßÔÚ¶ÁÈ¡Ò»¸öûÓÐÆ«ÒÆÁ¿µÄ·ÖÇø»òÕßÆ«ÒÆÁ¿ÎÞЧµÄÇé¿öϵĸÃÈçºÎ´¦Àí¡£ËüµÄĬÈÏÖµÊÇ latest£¬Òâ˼ָµÄÊÇ£¬ÔÚÆ«ÒÆÁ¿ÎÞЧµÄÇé¿öÏ£¬Ïû·ÑÕß½«´Ó×îеļǼ¿ªÊ¼¶ÁÈ¡Êý¾Ý¡£ÁíÒ»¸öÖµÊÇ earliest£¬Òâ˼ָµÄÊÇÔÚÆ«ÒÆÁ¿ÎÞЧµÄÇé¿öÏ£¬Ïû·ÑÕß½«´ÓÆðʼλÖô¦¿ªÊ¼¶ÁÈ¡·ÖÇøµÄ¼Ç¼¡£
• enable.auto.commit
  • 365betÌåÓýÔÚÏßÃÇÉԺ󽫽éÉܼ¸ÖÖ²»Í¬µÄÌύƫÒÆÁ¿µÄ·½Ê½¡£¸ÃÊôÐÔÖ¸¶¨ÁËÏû·ÑÕßÊÇ·ñ×Ô¶¯ÌύƫÒÆÁ¿£¬Ä¬ÈÏÖµÊÇ true£¬ÎªÁ˾¡Á¿±ÜÃâ³öÏÖÖظ´Êý¾ÝºÍÊý¾Ý¶ªÊ§£¬¿ÉÒÔ°ÑËüÉèÖÃΪ false£¬ÓÉ×Ô¼º¿ØÖƺÎʱÌύƫÒÆÁ¿¡£Èç¹û°ÑËüÉèÖÃΪ true£¬»¹¿ÉÒÔͨ¹ý auto.commit.interval.ms ÊôÐÔÀ´¿ØÖÆÌá½»µÄƵÂÊ
• partition.assignment.strategy
  • 365betÌåÓýÔÚÏßÃÇÖªµÀ£¬·ÖÇø»á·ÖÅä¸øȺ×éÖеÄÏû·ÑÕß¡£PartitionAssignor »á¸ù¾Ý¸ø¶¨µÄÏû·ÑÕߺÍÖ÷Ì⣬¾ö¶¨ÄÄЩ·ÖÇøÓ¦¸Ã±»·ÖÅä¸øÄĸöÏû·ÑÕߣ¬Kafka ÓÐÁ½¸öĬÈϵķÖÅä²ßÂÔRange ºÍ RoundRobin
• client.id
  • ¸ÃÊôÐÔ¿ÉÒÔÊÇÈ365betÌåÓýÔÚÏßâ×Ö·û´®£¬broker ÓÃËûÀ´±êʶ´Ó¿Í»§¶Ë·¢Ë͹ýÀ´µÄÏûÏ¢£¬Í¨³£±»ÓÃÔÚÈÕÖ¾¡¢¶ÈÁ¿Ö¸±êºÍÅä¶îÖÐ
• max.poll.records
  • ¸ÃÊôÐÔÓÃÓÚ¿ØÖƵ¥´Îµ÷Óà call() ·½·¨Äܹ»·µ»ØµÄ¼Ç¼ÊýÁ¿£¬¿ÉÒÔ°ïÄã¿ØÖÆÔÚÂÖѯÖÐÐèÒª´¦ÀíµÄÊý¾ÝÁ¿¡£
• receive.buffer.bytes ºÍ send.buffer.bytes
  • socket ÔÚ¶ÁдÊý¾ÝʱÓõ½µÄ TCP »º³åÇøÒ²¿ÉÒÔÉèÖôóС¡£Èç¹ûËüÃDZ»ÉèÖÃΪ -1£¬¾ÍʹÓòÙ×÷ϵͳĬÈÏÖµ¡£Èç¹ûÉú²úÕß»òÏû·ÑÕßÓë broker ´¦ÓÚ²»Í¬µÄÊý¾ÝÖÐÐÄÄÚ£¬¿ÉÒÔÊʵ±Ôö´óÕâЩֵ£¬ÒòΪ¿çÊý¾ÝÖÐÐĵÄÍøÂçÒ»°ã¶¼ÓбȽϸߵÄÑӳٺͱȽϵ͵Ĵø¿í¡£

 

Ìá½»ºÍÆ«ÒÆÁ¿µÄ¸ÅÄî

• ÌØÊâÆ«ÒÆ
  • 365betÌåÓýÔÚÏßÃÇÉÏÃæÌáµ½£¬Ïû·ÑÕßÔÚÿ´Îµ÷ÓÃpoll() ·½·¨½øÐж¨Ê±ÂÖѯµÄʱºò£¬»á·µ»ØÓÉÉú²úÕßдÈë Kafka µ«ÊÇ»¹Ã»Óб»Ïû·ÑÕßÏû·ÑµÄ¼Ç¼£¬Òò´Ë365betÌåÓýÔÚÏßÃÇ¿ÉÒÔ×·×Ùµ½ÄÄЩ¼Ç¼ÊDZ»Èº×éÀïµÄÄĸöÏû·ÑÕ߶ÁÈ¡µÄ¡£Ïû·ÑÕß¿ÉÒÔʹÓà Kafka À´×·×ÙÏûÏ¢ÔÚ·ÖÇøÖеÄλÖã¨Æ«ÒÆÁ¿£©
  • Ïû·ÑÕß»áÏòÒ»¸ö½Ð×ö _consumer_offset µÄÌØÊâÖ÷ÌâÖз¢ËÍÏûÏ¢£¬Õâ¸öÖ÷Ìâ»á±£´æÿ´ÎËù·¢ËÍÏûÏ¢ÖеķÖÇøÆ«ÒÆÁ¿£¬Õâ¸öÖ÷ÌâµÄÖ÷Òª×÷ÓþÍÊÇÏû·ÑÕß´¥·¢ÖØƽºâºó¼Ç¼ƫÒÆʹÓõģ¬Ïû·ÑÕßÿ´ÎÏòÕâ¸öÖ÷Ìâ·¢ËÍÏûÏ¢£¬Õý³£Çé¿öϲ»´¥·¢ÖØƽºâ£¬Õâ¸öÖ÷ÌâÊDz»Æð×÷Óõģ¬µ±´¥·¢ÖØƽºâºó£¬Ïû·ÑÕßÍ£Ö¹¹¤×÷£¬365ÌåÓýͶעÏû·ÑÕß¿ÉÄÜ»á·Öµ½¶ÔÓ¦µÄ·ÖÇø£¬Õâ¸öÖ÷Ìâ¾ÍÊÇÈÃÏû·ÑÕßÄܹ»¼ÌÐø´¦ÀíÏûÏ¢ËùÉèÖõġ£
  • Èç¹ûÌá½»µÄÆ«ÒÆÁ¿Ð¡ÓÚ¿Í»§¶Ë×îºóÒ»´Î´¦ÀíµÄÆ«ÒÆÁ¿£¬ÄÇôλÓÚÁ½¸öÆ«ÒÆÁ¿Ö®¼äµÄÏûÏ¢¾Í»á±»Öظ´´¦Àí

• Èç¹ûÌá½»µÄÆ«ÒÆÁ¿´óÓÚ×îºóÒ»´ÎÏû·ÑʱµÄÆ«ÒÆÁ¿£¬ÄÇô´¦ÓÚÁ½¸öÆ«ÒÆÁ¿ÖмäµÄÏûÏ¢½«»á¶ªÊ§

• ¼ÈÈ»_consumer_offset Èç´ËÖØÒª£¬ÄÇôËüµÄÌá½»·½Ê½ÊÇÔõÑùµÄÄØ£¿ÏÂÃæ365betÌåÓýÔÚÏßÃǾÍÀ´ËµÒ»ÏÂ####Ìá½»·½Ê½
• KafkaConsumer API ÌṩÁ˶àÖÖ·½Ê½À´ÌύƫÒÆÁ¿
• ×Ô¶¯Ìá½»
  • ×î¼òµ¥µÄ·½Ê½¾ÍÊÇÈÃÏû·ÑÕß×Ô¶¯ÌύƫÒÆÁ¿¡£Èç¹û enable.auto.commit ±»ÉèÖÃΪtrue£¬ÄÇôÿ¹ý 5s£¬Ïû·ÑÕß»á×Ô¶¯°Ñ´Ó poll() ·½·¨ÂÖѯµ½µÄ×î´óÆ«ÒÆÁ¿Ìá½»ÉÏÈ¥¡£Ìύʱ¼ä¼ä¸ôÓÉ auto.commit.interval.ms ¿ØÖÆ£¬Ä¬ÈÏÊÇ 5s¡£ÓëÏû·ÑÕßÀïµÄÆäËû¶«Î÷Ò»Ñù£¬×Ô¶¯Ìá½»Ò²ÊÇÔÚÂÖѯÖнøÐеġ£Ïû·ÑÕßÔÚÿ´ÎÂÖѯÖлá¼ì²éÊÇ·ñÌá½»¸ÃÆ«ÒÆÁ¿ÁË£¬Èç¹ûÊÇ£¬ÄÇô¾Í»áÌá½»´ÓÉÏÒ»´ÎÂÖѯÖзµ»ØµÄÆ«ÒÆÁ¿¡£
• Ìá½»µ±Ç°Æ«ÒÆÁ¿
  • °Ñ auto.commit.offset ÉèÖÃΪ false£¬¿ÉÒÔÈÃÓ¦ÓóÌÐò¾ö¶¨ºÎʱÌύƫÒÆÁ¿¡£Ê¹Óà commitSync() ÌύƫÒÆÁ¿¡£Õâ¸ö API »áÌá½»ÓÉ poll() ·½·¨·µ»ØµÄ×îÐÂÆ«ÒÆÁ¿£¬Ìá½»³É¹¦ºóÂíÉÏ·µ»Ø£¬Èç¹ûÌύʧ°Ü¾ÍÅ׳öÒì³£¡£
  • commitSync() ½«»áÌá½»ÓÉ poll() ·µ»ØµÄ×îÐÂÆ«ÒÆÁ¿£¬Èç¹û´¦ÀíÍêËùÓмǼºóҪȷ±£µ÷ÓÃÁË commitSync()£¬·ñÔò»¹ÊÇ»áÓжªÊ§ÏûÏ¢µÄ·çÏÕ£¬Èç¹û·¢ÉúÁËÔÚ¾ùºâ£¬´Ó×î½üÒ»ÅúÏûÏ¢µ½·¢ÉúÔÚ¾ùºâÖ®¼äµÄËùÓÐÏûÏ¢¶¼½«±»Öظ´´¦Àí¡£
• Òì²½Ìá½»
  • Òì²½Ìá½» commitAsync() Óëͬ²½Ìá½» commitSync() ×î´óµÄÇø±ðÔÚÓÚÒì²½Ìá½»²»»á½øÐÐÖØÊÔ£¬Í¬²½Ìá½»»áÒ»Ö½øÐÐÖØÊÔ¡£
• ͬ²½ºÍÒì²½×éºÏÌá½»
  • Ò»°ãÇé¿öÏ£¬Õë¶Ôż¶û³öÏÖµÄÌύʧ°Ü£¬²»½øÐÐÖØÊÔ²»»áÓÐÌ«´óµÄÎÊÌ⣬ÒòΪÈç¹ûÌύʧ°ÜÊÇÒòΪÁÙʱÎÊÌâµ¼Öµģ¬ÄÇôºóÐøµÄÌá½»×Ü»áÓгɹ¦µÄ¡£µ«ÊÇÈç¹ûÔڹرÕÏû·ÑÕß»òÔÙ¾ùºâÇ°µÄ×îºóÒ»´ÎÌá½»£¬¾ÍҪȷ±£Ìá½»³É¹¦¡£
  • Òò´Ë£¬ÔÚÏû·ÑÕ߹رÕ֮ǰһ°ã»á×éºÏʹÓÃcommitAsyncºÍcommitSyncÌύƫÒÆÁ¿¡£
• Ìá½»Ìض¨µÄÆ«ÒÆÁ¿
  • Ïû·ÑÕßAPIÔÊÐíµ÷Óà commitSync() ºÍ commitAsync() ·½·¨Ê±´«ÈëÏ£ÍûÌá½»µÄ partition ºÍ offset µÄ map£¬¼´Ìá½»Ìض¨µÄÆ«ÒÆÁ¿¡£

---

 

• ¡¶Kafka ȨÍþÖ¸ÄÏ-https://kafka.apache.org/¡·
• ¡¶Kafka ȨÍþÖ¸ÄÏ-https://dzone.com/articles/what-is-kafka¡·
• ¡¶Kafka ȨÍþÖ¸ÄÏ-http://kafka.apache.org/documentation/¡·
• ¡¶Kafka ȨÍþÖ¸ÄÏ-https://www.tutorialkart.com/apache-kafka-tutorial/¡·

---

¹Ø×¢¡¢ÁôÑÔ£¬365betÌåÓýÔÚÏßÃÇÒ»Æðѧϰ£¬ÄúµÄÊÕ²ØÊÇ365betÌåÓýÔÚÏß³ÖÐø¸üеĶ¯Á¦£¡

 

===============Talk is cheap, show me the code,bye-bye================