当前位置： 首页 > news >正文

北京企业网站推广网上怎么找人去推广广告

news 2025/7/12 19:16:37
北京企业网站推广,网上怎么找人去推广广告,个人备案的网站销售商品,做任务送科比网站目 录 1. 新建一个消息生产者2. 新建一个消息消费者3. 测 试 在开始之前，需要先做点准备工作，用 IDEA 新建一个 Maven 项目，取名 kafka-study，然后删掉它的 src 目录，接着在 pom.xml 里面引入下面的依赖。这个项目的作…

目 录

    • 1. 新建一个消息生产者
    • 2. 新建一个消息消费者
    • 3. 测 试

在开始之前，需要先做点准备工作，用 IDEA 新建一个 Maven 项目，取名 kafka-study，然后删掉它的 src 目录，接着在 pom.xml 里面引入下面的依赖。这个项目的作用是作为整个工程的父项目，后面的项目都在此基础上新建 module 即可。

这里用到的环境信息如下：

  • JDK1.8
  • SpringBoot2.7.1
  • IDEA
  • Maven3.6.3
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"><modelVersion>4.0.0</modelVersion><packaging>pom</packaging><parent><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-parent</artifactId><version>2.7.1</version><relativePath/></parent><groupId>org.yuhuofei</groupId><artifactId>kafka-study</artifactId><version>1.0-SNAPSHOT</version><name>kafka-study</name><description>Study project for Kafka</description><properties><java.version>1.8</java.version></properties><dependencies><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-web</artifactId></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-test</artifactId><scope>test</scope></dependency></dependencies><build><plugins><plugin><groupId>org.springframework.boot</groupId><artifactId>spring-boot-maven-plugin</artifactId></plugin></plugins></build>
</project>

1. 新建一个消息生产者

使用 IDEA 在 kafka-study 的基础上新建一个 Maven 类型的 module ，命名为 producer-01，直至完成。接着，我们改造这个 module ，把它变成我们想要的消息生产者。

1、在 java 目录下，新建一个包 com.yuhuofei ，然后再创建一个启动类 ProducerApplication.java ，内容如下:

package com.yuhuofei;import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;@SpringBootApplication
public class ProducerApplication {public static void main(String[] args) {SpringApplication.run(ProducerApplication.class, args);}}

2、在 resources 目录下，新建配置文件 application.properties ，内容如下：

server.port=8080
server.servlet.context-path=/producer-01
#Kafka配置
topic.name=my-kafka-topic

3、在自己的 pom.xml 文件中，引入 Kafka 客户端的依赖。这里需要注意的是，由于我们在上一篇博客里安装的是 kafka_2.11-2.2.1 版本的服务端，所以引入的 Kafka 客户端的版本最好对应，也选择 2.2.1 版本的。

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"><parent><artifactId>kafka-study</artifactId><groupId>org.yuhuofei</groupId><version>1.0-SNAPSHOT</version></parent><modelVersion>4.0.0</modelVersion><artifactId>producer-01</artifactId><dependencies><!--官方原生的java版kafka客户端依赖--><dependency><groupId>org.apache.kafka</groupId><artifactId>kafka-clients</artifactId><version>2.2.1</version></dependency><!--fastjson依赖--><dependency><groupId>com.alibaba</groupId><artifactId>fastjson</artifactId><version>2.0.11.graal</version></dependency></dependencies></project>

4、由于 Kafka 进行消息的发布或者消费是需要有 Topic 的，因此我们需要要创建一个 Topic。在 zookeeper 和 Kafka 服务端都启动的情况下，打开 D:\Kafka\kafka_2.11-2.2.1\bin\windows 文件夹，在地址栏输入 cmd，打开控制台，然后用下面的命令创建、查看、删除 topic 。

如下图所示，我已经创建了一个名字为 my-kafka-topic 的 topic ，这个 topic 的分区为 1，副本也是 1。

# 创建topic
kafka-topics.bat --create --zookeeper 127.0.0.1:2181 --replication-factor 1 --partitions 1 --topic my-kafka-topic
# 查询topic列表
kafka-topics.bat --zookeeper 127.0.0.1:2181 --list					
# 删除topic
kafka-topics.bat --delete --zookeeper 127.0.0.1:2181 --topic my-kafka-topic
# 查询某个topic信息
kafka-topics.bat --describe --zookeeper localhost:2181 --topic my-kafka-topic
#kafka查看topic数据内容
kafka-console-consumer.bat --bootstrap-server localhost:9092 --topic my-kafka-topic --from-beginning

在这里插入图片描述

5、实现生产者发送消息

  • controller层
package com.yuhuofei.controller;import com.yuhuofei.service.ProducerInterface;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;import javax.annotation.Resource;/*** @Description* @ClassName ProducerController* @Author yuhuofei* @Date 2023/8/13 16:38* @Version 1.0*/
@RequestMapping("/info")
@RestController
public class ProducerController {@Resourceprivate ProducerInterface producerInterface;@GetMapping("/create-message")public void createTopic() {producerInterface.createMessage();}
}
  • service层
package com.yuhuofei.service;/*** @Description* @InterfaceName ProducerController* @Author yuhuofei* @Date 2023/8/13 16:39* @Version 1.0*/
public interface ProducerInterface {void createMessage();
}
  • serviceImpl层
package com.yuhuofei.service.impl;import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.yuhuofei.service.ProducerInterface;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;import javax.annotation.Resource;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;/*** @Description* @ClassName ProducerInterfaceImpl* @Author yuhuofei* @Date 2023/8/13 16:39* @Version 1.0*/
@Service
public class ProducerInterfaceImpl implements ProducerInterface {@Value("${topic.name}")private String topicName;@Resourceprivate Producer producer;@Overridepublic void createMessage() {//创建测试数据JSONObject jsonObject = new JSONObject();jsonObject.put("word", "这是生产者生产的测试消息");String jsonString = JSON.toJSONString(jsonObject);//使用kafka发送消息ProducerRecord<String, String> producerRecord = new ProducerRecord<>(topicName, jsonString);producer.send(producerRecord);}
}
  • 客户端配置类
package com.yuhuofei.config;import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.StringSerializer;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;import java.util.Properties;/*** @Description 配置生产者客户端参数* @ClassName KafkaProperties* @Author yuhuofei* @Date 2023/8/13 22:18* @Version 1.0*/
@Configuration
public class KafkaConfig {@Bean("producer")public Producer<String, String> getKafkaProducer() {Properties properties = new Properties();//配置Kafka的服务端IP和端口properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "127.0.0.1:9092");//指定key使用的序列化类properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());//指定value使用的序列化类properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());return new KafkaProducer<>(properties);}
}

6、启动服务，调用接口 http://localhost:8080/producer-01/info/create-message ，然后在控制台用命令查看一下消息是否写入，如下图所见，虽然是乱码，但是已经可以看出消息成功写入到 Kafka

在这里插入图片描述
到这里，消息生产者的创建暂时告一段落。

2. 新建一个消息消费者

和前面建立生产者类似的步骤，只是文件信息和名字不一样，这里步骤就不重复了，直接给出相关的配置和类。这个消费者，取名 consumer-01 。

  • pom.xml文件
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"><parent><artifactId>kafka-study</artifactId><groupId>org.yuhuofei</groupId><version>1.0-SNAPSHOT</version></parent><modelVersion>4.0.0</modelVersion><artifactId>consumer-01</artifactId><dependencies><!--官方原生的java版kafka客户端依赖--><dependency><groupId>org.apache.kafka</groupId><artifactId>kafka-clients</artifactId><version>2.2.1</version></dependency><!--fastjson依赖--><dependency><groupId>com.alibaba</groupId><artifactId>fastjson</artifactId><version>2.0.11.graal</version></dependency></dependencies>
</project>
  • properties文件
server.port=8081
server.servlet.context-path=/consumer-01
#Kafka配置
topic.name=my-kafka-topic
  • 启动类 ConsumerApplication.java
package com.yuhuofei;import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;@SpringBootApplication
public class ConsumerApplication {public static void main(String[] args) {SpringApplication.run(ConsumerApplication.class, args);}}
  • 控制层ConsumerController.java
package com.yuhuofei.controller;import com.yuhuofei.service.ConsumerInterface;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;import javax.annotation.Resource;/*** @Description* @ClassName ConsumerController* @Author yuhuofei* @Date 2023/8/14 0:19* @Version 1.0*/
@RestController
@RequestMapping("/consumer")
public class ConsumerController {@Resourceprivate ConsumerInterface consumerInterface;@GetMapping("/getData")public void getData() {consumerInterface.getData();}
}
  • service层
package com.yuhuofei.service;/*** @Description* @ClassName ConsumerInterface* @Author yuhuofei* @Date 2023/8/14 0:20* @Version 1.0*/
public interface ConsumerInterface {void getData();
}
  • serviceImpl层
package com.yuhuofei.service.impl;import com.yuhuofei.service.ConsumerInterface;
import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;import javax.annotation.Resource;
import java.time.Duration;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.List;/*** @Description* @ClassName ConsumerInterfaceImpl* @Author yuhuofei* @Date 2023/8/14 0:20* @Version 1.0*/
@Service
public class ConsumerInterfaceImpl implements ConsumerInterface {@Value("${topic.name}")private String topicName;@Resourceprivate Consumer consumer;@Overridepublic void getData() {Collection<String> topics = Collections.singletonList(topicName);consumer.subscribe(topics);List<String> result = new ArrayList<>();while (true) {ConsumerRecords<String, String> consumerRecords = consumer.poll(Duration.ofMillis(1000));for (ConsumerRecord consumerRecord : consumerRecords) {//从ConsumerRecord中获取消费数据String res = (String) consumerRecord.value();System.out.println("从Kafka中消费的原始数据: " + res);result.add(res);System.out.println(result);}}}
}
  • config配置类
package com.yuhuofei.config;import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.kafka.common.serialization.StringSerializer;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;import java.util.Properties;/*** @Description 配置消费者参数* @ClassName KafkaConsumerConfig* @Author yuhuofei* @Date 2023/8/14 0:08* @Version 1.0*/
@Configuration
public class KafkaConsumerConfig {@Bean("consumer")public Consumer<String, String> getKafkaConsumer() {Properties properties = new Properties();//配置Kafka的服务端IP和端口properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "127.0.0.1:9092");//指定key使用的序列化类properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());//指定value使用的序列化类properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());//指定消费者组properties.put(ConsumerConfig.GROUP_ID_CONFIG, "consumer-01-group");properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG,true);properties.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG,100);properties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,"latest");return new KafkaConsumer<>(properties);}
}

最终整个项目目录如下图所示。

在这里插入图片描述

3. 测 试

1、检查 zookeeper 和 Kafka 服务端是否是启动的，没启动就先启动。

2、接着，启动消费者服务，然后调用一下接口 http://localhost:8081/consumer-01/consumer/getData ，这么做的目的启动一个线程，监听 topic 。

3、最后，启动生产者服务，调用接口 http://localhost:8080/producer-01/info/create-message ，向 Kafka 服务器写数据。

结果如下所示：

控制台看到的所有消息记录
这是所有发送的记录
生产者发送一条消息
在这里插入图片描述
消费者监听到一条消息并接收
在这里插入图片描述

在 SpringBoot 中，使用官方原生 java 版 Kafka 客户端的入门介绍到这里结束，还有更多使用方式，等待挖掘。

http://www.ritt.cn/news/25279.html

相关文章：

  • 网站开发 ssh 菜鸟优化网站页面
  • 网站建设及服务合同书引擎搜索是什么意思
  • 可信网站认证必须做吧免费二级域名申请网站
  • net网站开发参考文献站长域名查询
  • 网站建设规划公司地址磁力兔子
  • 怎么做网站数据库近期重大新闻
  • 做b2b2c商城网站杭州网站推广找哪家
  • 天津微网站建设百度app官方下载
  • 网站建设方案.docsemir是什么牌子
  • 律师个人 网站做优化百度文库个人登录
  • 无锡做食品网站的公司哪家好四川seo
  • 知名网站建设公司免费代码网站
  • 做自己的网站要钱么网络推广服务外包
  • 珠海网络公司排名搜索引擎优化seo应用
  • 学做电商的网站有哪些企业seo网络推广
  • 自己建设网站的利弊免费访问国外网站的app
  • 公司名字大全推荐单页网站排名优化
  • 站长网站被跳转怎么办西安今日头条新闻消息
  • 网络规划设计师考纲教材改版新旧对比免费seo网站的工具
  • wordpress修改文章时间股票发行ipo和seo是什么意思
  • qq手机版福建seo推广方案
  • 做企业网站 需要用服务器吗百度平台商家我的订单查询
  • 做游戏课程网站成都竞价托管多少钱
  • 温州网站制作系统韩国热搜榜
  • 官网小程序优化网站推广
  • 网站建设与管理内容seo技术外包公司
  • 网页设计入门 电子书下载搜索引擎的关键词优化
  • 网上做服装批发网站网络推广软文怎么写
  • 如何构建个人网站产品推广公司
  • html5高端网站建设温州免费建站模板