devbox
我家自动化
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

kafka API练习_scala.actors.threadpool pom

作者：我家自动化 | 2024-08-22 20:03:41

赞

踩

scala.actors.threadpool pom

生产者层面

  1. package Kafka
  2.  
  3.  
  4. import java.util.Properties
  5.  
  6. import kafka.producer.{KeyedMessage, Producer, ProducerConfig}
  7.  
  8.  
  9. object KafkaProducerTest {
  10.   def main(args: Array[String]): Unit = {
  11.     //定义topic，把数据传到该topic
  12.     val topic = "KafkaSimple3"
  13.     //创建一个配置文件信息类
  14.     val properties = new Properties()
  15.     //数据在序列化的编码类型
  16.     properties.put("serializer.class", "kafka.serializer.StringEncoder")
  17.     //kafaka集群列表
  18.     properties.put("metadata.broker.list", "192.168.42.132:9092,192.168.42.134:9092,192.168.42.135:9092")
  19.     //设置发送数据后是否需要服务端的反馈:0,1,-1
  20.     properties.put("request.required.acks", "1")
  21.     //调用分区器
  22.     properties.put("partitioner.class", "Kafka.CustomPartitioner")
  23.  
  24.     val config: ProducerConfig = new ProducerConfig(properties)
  25.     //创建一个生产者实例vb
  26.     val producer: Producer[String, String] = new Producer(config)
  27.  
  28.     //模拟生产一些数据
  29.     for (i <- 1 to 10000) {
  30.       val msg = s"$i: Producer send data"
  31.       producer.send(new KeyedMessage[String, String](topic, msg))
  32.     }
  33.  
  34.   }
  35.  
  36. }

消费者层面

  1. package Kafka
  2.  
  3. import java.util.Properties
  4.  
  5. import kafka.consumer._
  6. import kafka.message.MessageAndMetadata
  7.  
  8. import scala.actors.threadpool.{ExecutorService, Executors}
  9. import scala.collection.mutable
  10.  
  11. class KafkaConsumer(val consumer: String, val stream: KafkaStream[Array[Byte], Array[Byte]]) extends Runnable {
  12.   override def run(): Unit = {
  13.     val it: ConsumerIterator[Array[Byte], Array[Byte]] = stream.iterator()
  14.     while (it.hasNext()) {
  15.       val data: MessageAndMetadata[Array[Byte], Array[Byte]] = it.next()
  16.       val topic: String = data.topic
  17.       val partition: Int = data.partition
  18.       val offset: Long = data.offset
  19.       val msg = new String(data.message())
  20.       println(s"Consumer: $consumer, Topic: $topic, Partition: $partition, Offset: $offset, msg: $msg")
  21.     }
  22.   }
  23. }
  24.  
  25. object KafkaConsumerTest {
  26.   def main(args: Array[String]): Unit = {
  27.     //定义用来读取数据的topic
  28.     val topic = "KafkaSimple3"
  29.  
  30.     //用来存储多个topic
  31.     val topics = new mutable.HashMap[String, Int]()
  32.     topics.put(topic, 2)
  33.  
  34.     //配置文件信息
  35.     val properties = new Properties()
  36.     //ConsumerGroup id
  37.     properties.put("group.id", "zwj-consumer-group")
  38.     // 指定zookeeper的地址列表, 注意：value里不要有空格
  39.     properties.put("zookeeper.connect", "192.168.42.132:2181,192.168.42.134:2181,192.168.42.135:2181")
  40.     // 如果zookeeper没有offset值或者offset值超出范围，需要指定一个初始的offset
  41.     properties.put("auto.offset.reset", "smallest")
  42.     //把配置信息封装到ConsumerConfig对象里
  43.     val config = new ConsumerConfig(properties)
  44.  
  45.     //创建Consumer实例，如果没有数据，会一直线程等待
  46.     val consumer: ConsumerConnector = Consumer.create(config)
  47.  
  48.     //根据所传topics来获取数据，得到一个stream流
  49.     val streams: collection.Map[String, List[KafkaStream[Array[Byte], Array[Byte]]]] = consumer.createMessageStreams(topics)
  50.  
  51.     //获取指定topic的数据
  52.     val stream: Option[List[KafkaStream[Array[Byte], Array[Byte]]]] = streams.get(topic)
  53.  
  54.     //创建一个固定大小的线程池
  55.     val pool: ExecutorService = Executors.newFixedThreadPool(3)
  56.     for (i <- 0 until stream.size) {
  57.       pool.execute(new KafkaConsumer(s"Consumer: $i", stream.get(i)))
  58.     }
  59.  
  60.  
  61.   }
  62. }

pom.xml（里面有比较多不必要的包是写之前的代码残留的）

  1. <?xml version="1.0" encoding="UTF-8"?>
  2. <project xmlns="http://maven.apache.org/POM/4.0.0"
  3.          xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
  4.          xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
  5.     <modelVersion>4.0.0</modelVersion>
  6.  
  7.     <groupId>com.zwj</groupId>
  8.     <artifactId>ScalaTest</artifactId>
  9.     <version>1.0-SNAPSHOT</version>
  10.     <properties>
  11.         <maven.compiler.source>1.7</maven.compiler.source>
  12.         <maven.compiler.target>1.7</maven.compiler.target>
  13.         <encoding>UTF-8</encoding>
  14.         <scala.version>2.10.6</scala.version>
  15.         <spark.version>1.6.3</spark.version>
  16.         <hadoop.version>2.6.4</hadoop.version>
  17.     </properties>
  18.  
  19.     <dependencies>
  20.         <dependency>
  21.             <groupId>org.scala-lang</groupId>
  22.             <artifactId>scala-library</artifactId>
  23.             <version>${scala.version}</version>
  24.         </dependency>
  25.  
  26.         <dependency>
  27.             <groupId>org.apache.spark</groupId>
  28.             <artifactId>spark-core_2.10</artifactId>
  29.             <version>${spark.version}</version>
  30.         </dependency>
  31.  
  32.         <dependency>
  33.             <groupId>org.apache.hadoop</groupId>
  34.             <artifactId>hadoop-client</artifactId>
  35.             <version>${hadoop.version}</version>
  36.         </dependency>
  37.         <!-- https://mvnrepository.com/artifact/org.apache.kafka/kafka -->
  38.         <!-- https://mvnrepository.com/artifact/org.apache.kafka/kafka -->
  39.         <!-- https://mvnrepository.com/artifact/org.apache.kafka/kafka -->
  40.         <dependency>
  41.             <groupId>org.apache.kafka</groupId>
  42.             <artifactId>kafka_2.10</artifactId>
  43.             <version>0.9.0.1</version>
  44.         </dependency>
  45.  
  46.         <!-- https://mvnrepository.com/artifact/org.apache.kafka/kafka-clients -->
  47.         <dependency>
  48.             <groupId>org.apache.kafka</groupId>
  49.             <artifactId>kafka-clients</artifactId>
  50.             <version>0.9.0.1</version>
  51.         </dependency>
  52.  
  53.         <!-- https://mvnrepository.com/artifact/log4j/log4j -->
  54.         <dependency>
  55.             <groupId>org.apache.logging.log4j</groupId>
  56.             <artifactId>log4j-core</artifactId>
  57.             <version>2.10.0</version>
  58.         </dependency>
  59.  
  60.  
  61.         <!-- https://mvnrepository.com/artifact/org.apache.kafka/kafka-streams -->
  62.         <dependency>
  63.             <groupId>org.apache.kafka</groupId>
  64.             <artifactId>kafka-streams</artifactId>
  65.             <version>0.10.0.1</version>
  66.         </dependency>
  67.  
  68.         <dependency>
  69.             <groupId>org.apache.kafka</groupId>
  70.             <artifactId>connect-api</artifactId>
  71.             <version>0.10.0.1</version>
  72.         </dependency>
  73.  
  74.         <!-- https://mvnrepository.com/artifact/org.apache.spark/spark-sql -->
  75.         <dependency>
  76.             <groupId>org.apache.spark</groupId>
  77.             <artifactId>spark-sql_2.10</artifactId>
  78.             <version>1.6.3</version>
  79.         </dependency>
  80.         <!-- https://mvnrepository.com/artifact/mysql/mysql-connector-java -->
  81.         <dependency>
  82.             <groupId>mysql</groupId>
  83.             <artifactId>mysql-connector-java</artifactId>
  84.             <version>5.1.46</version>
  85.         </dependency>
  86.  
  87.     </dependencies>
  88.  
  89.     <build>
  90.         <sourceDirectory>src/main/scala</sourceDirectory>
  91.         <testSourceDirectory>src/test/scala</testSourceDirectory>
  92.         <plugins>
  93.             <plugin>
  94.                 <groupId>net.alchim31.maven</groupId>
  95.                 <artifactId>scala-maven-plugin</artifactId>
  96.                 <version>3.2.2</version>
  97.                 <executions>
  98.                     <execution>
  99.                         <goals>
  100.                             <goal>compile</goal>
  101.                             <goal>testCompile</goal>
  102.                         </goals>
  103.                         <configuration>
  104.                             <args>
  105.                                 <arg>-make:transitive</arg>
  106.                                 <arg>-dependencyfile</arg>
  107.                                 <arg>${project.build.directory}/.scala_dependencies</arg>
  108.                             </args>
  109.                         </configuration>
  110.                     </execution>
  111.                 </executions>
  112.             </plugin>
  113.  
  114.             <plugin>
  115.  
  116.                 <groupId>org.apache.maven.plugins</groupId>
  117.                 <artifactId>maven-shade-plugin</artifactId>
  118.                 <version>2.4.3</version>
  119.                 <executions>
  120.                     <execution>
  121.                         <phase>package</phase>
  122.                         <goals>
  123.                             <goal>shade</goal>
  124.                         </goals>
  125.                         <configuration>
  126.                             <filters>
  127.                                 <filter>
  128.                                     <artifact>*:*</artifact>
  129.                                     <includes>
  130.                                         <include>META-INF/*.SF</include>
  131.                                         <include>META-INF/*.DSA</include>
  132.                                         <include>META-INF/*.RSA</include>
  133.                                         <include>META-INF/*.MF</include>
  134.                                     </includes>
  135.                                     <excludes>
  136.                                         <exclude>META-INF/*</exclude>
  137.                                     </excludes>
  138.                                 </filter>
  139.                             </filters>
  140.                             <transformers>
  141.                                 <transformer
  142.                                         implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer">
  143.                                     <mainClass>com.qf.spark.WordCount</mainClass>
  144.                                 </transformer>
  145.                             </transformers>
  146.                         </configuration>
  147.                     </execution>
  148.                 </executions>
  149.             </plugin>
  150.         </plugins>
  151.     </build>
  152. </project>

 

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/我家自动化/article/detail/1017792
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号