Linux中nacos的拉取安装
拉取naocs镜像：docker pull nacos/nacos-server:1.2.0
创建容器：docker run --env MODE=standalone --name nacos --restart=always -d -p 8848:8848 nacos/nacos-server:1.2.0
访问地址：http://192.168.200.130:8848/nacos nacos网关分析
注册中心gateway
server:
port: 51601
spring:
application:
name: leadnews-app-gateway
cloud:
nacos:
discovery:
server-addr: 192.168.74.128:8848
config:
server-addr: 192.168.74.128:8848
file-extension: ymlspring:
cloud:
gateway:
globalcors:
add-to-simple-url-handler-mapping: true
corsConfigurations:
'[/**]':
allowedHeaders: "*"
allowedOrigins: "*"
allowedMethods:
- GET
- POST
- DELETE
- PUT
- OPTION
routes:
# 平台管理
- id: user
uri: lb://leadnews-user
predicates:
- Path=/user/**
filters:
- StripPrefix= 1分析：
这个配置项是Spring Cloud Gateway的路由配置，表示路由规则。
routes: [ { id: user, uri: lb://leadnews-user, predicates: [ Path=/user/** ], filters: [ StripPrefix=1 ] } ]，这个配置项是一个路由规则，表示当请求的路径以"/user/"开头时，将请求转发到uri为lb://leadnews-user的微服务，
stripPrefix=1表示在转发请求时，将路径中的第一个"/user/"前缀去除。
lead_news_user
server:
port: 51601
spring:
application:
name: leadnews-app-gateway
cloud:
nacos:
discovery:
server-addr: 192.168.74.128:8848
config:
server-addr: 192.168.74.128:8848
file-extension: ymlnginx的配置分析
#user nobody;
worker_processes 1;
events {
worker_connections 1024;
}
http {
include mime.types;
default_type application/octet-stream;
sendfile on;
keepalive_timeout 65;
# 引入自定义配置文件
include leadnews.conf/*.conf;
}conf分析：
1.
这个配置文件是一个Nginx主配置文件，用于配置Nginx的运行环境和代理设置。
#user nobody;，这个配置项表示Nginx运行用户为nobody。
worker_processes 1;，这个配置项表示Nginx的工作进程数为1。
events { worker_connections 1024; }，这个配置项是Nginx的事件配置，表示Nginx每个工作进程的最大连接数为1024。
http { include mime.types; default_type application/octet-stream; sendfile on; keepalive_timeout 65; # 引入自定义配置文件 include leadnews.conf/*.conf; }，这个配置项是Nginx的HTTP服务配置，表示Nginx支持的MIME类型，以及默认的内容类型，启用sendfile特性，超时时间为65秒，引入自定义配置文件leadnews.conf下的所有配置文件。
leadnews.conf->heima-leadnews-app.conf 分析
upstream heima-app-gateway{
server localhost:51601;
}
server {
listen 8801;
location / {
root D:/Idea_Project/SpringCloud/toutaio/app-web/;
index index.html;
}
location ~/app/(.*) {
proxy_pass http://heima-app-gateway/$1;
proxy_set_header HOST $host; # 不改变源请求头的值
proxy_pass_request_body on; #开启获取请求体
proxy_pass_request_headers on; #开启获取请求头
proxy_set_header X-Real-IP $remote_addr; # 记录真实发出请求的客户端IP
proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for; #记录代理信息
}
}分析：
这个配置文件是一个Nginx服务器的配置文件，用于代理请求到Spring Cloud Gateway。 upstream heima-app-gateway{ server localhost:51601; }，这个配置项是上游服务器的配置，表示Nginx将请求代理到localhost主机的51601端口。
这个配置项是Nginx的服务器配置，表示Nginx监听8801端口。 location / { root D:/Idea_Project/SpringCloud/toutaio/app-web/; index index.html; }，这个配置项是Nginx的location配置，表示当接收到请求时，如果请求的路径与正则表达式匹配，则进行相应的处理。在这个配置项中，location的正则表达式为"/"，表示匹配所有路径。root D:/Idea_Project/SpringCloud/toutaio/app-web/;表示请求的文件根目录为D:/Idea_Project/SpringCloud/toutaio/app-web/。index index.html;表示当请求的路径没有文件名时，返回index.html文件。
Freemarker
freemarker作为springmvc一种视图格式，默认情况下SpringMVC支持freemarker视图格式。
可以制作模板文件，输出成html文件。
@Test
public void test() throws IOException, TemplateException {
//freemarker的模板对象，获取模板
Template template = configuration.getTemplate("02-list.ftl");
Map params = getData();
//合成
//第一个参数 数据模型
//第二个参数 输出流
template.process(params, new FileWriter("d:/list.html"));
}
Minio(文件分布式系统)
简介
MinIO基于Apache License v2.0开源协议的对象存储服务，可以做为云存储的解决方案用来保存海量的图片，视频，文档。由于采用Golang实现，服务端可以工作在Windows,Linux, OS X和FreeBSD上。配置简单，基本是复制可执行程序，单行命令可以运行起来。
安装
拉取稳定版本:docker pull minio/minio:RELEASE.2021-06-17T00-10-46Z-28-gac7697426
创建并运行容器： docker run -d -p 9000:9000 --name minio \
-e "MINIO_ACCESS_KEY=minio" \
-e "MINIO_SECRET_KEY=minio123" \
-v /path/to/data:/data \
-v /path/to/config:/root/.minio \
minio/minio:RELEASE.2021-06-17T00-10-46Z server /data
访问minmo系统： http://192.168.74.128:9000手动上传文件到minio
public static void main(String[] args) {
FileInputStream fileInputStream = null;
try {
fileInputStream = new FileInputStream("e:\\index.js");;
//1.创建minio链接客户端
MinioClient minioClient = MinioClient.builder().credentials("minio", "minio123").endpoint("http://192.168.74.128:9000").build();
//2.上传
PutObjectArgs putObjectArgs = PutObjectArgs.builder()
.object("plugins/js/index.js")//文件名
.contentType("text/js")//文件类型
.bucket("leadnews")//桶名词 与minio创建的名词一致
.stream(fileInputStream, fileInputStream.available(), -1) //文件流
.build();
minioClient.putObject(putObjectArgs);
} catch (Exception ex) {
ex.printStackTrace();
}
}延迟任务
什么是延迟任务？
定时任务：有固定周期的，有明确的触发事件
延迟任务：没有固定的开始时间，常常是由一个事件触发的，而在这个事件触发之后的一段事件内触发另一个事件，任务可以立即执行，也可以延迟
延迟队列服务
悲观锁
每次去拿数据的时候都会认为别人会修改，所以每次在拿数据的时候都会上锁。
乐观锁
每次去拿数据的时候都会认为别人不会修改，所以不会上锁，但是在更新的时候会判断一下在此期间别人有没有去更新这个数据，可以使用版本号等机制。
乐观锁的集成
启动类添加
package com.heima.schedule;
import com.baomidou.mybatisplus.extension.plugins.MybatisPlusInterceptor;
import com.baomidou.mybatisplus.extension.plugins.inner.OptimisticLockerInnerInterceptor;
import org.mybatis.spring.annotation.MapperScan;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.annotation.Bean;
@SpringBootApplication
@MapperScan("com.heima.schedule.mapper")
public class ScheduleApplication {
public static void main(String[] args) {
SpringApplication.run(ScheduleApplication.class,args);
}
/**
* mybatis-plus乐观锁支持
* @return
*/
@Bean
public MybatisPlusInterceptor optimisticLockerInterceptor(){
MybatisPlusInterceptor interceptor = new MybatisPlusInterceptor();
interceptor.addInnerInterceptor(new OptimisticLockerInnerInterceptor());
return interceptor;
}
}
安装docker redis镜像
拉取镜像
docker pull redis创建容器
docker run -d --name redis --restart=always -p 6379:6379 redis --requirepass "leadnews"连接测试
分布式锁解决集群下的方法抢占执行——分布式锁
场景：两台服务器，执行一个定时任务
分布式锁：控制分布式系统有序的去对共享资源进行操作，通过互斥来保证数据的一致性
分布式锁的解决方案：
加锁：
/**
* 加锁
*
* @param name
* @param expire
* @return
*/
public String tryLock(String name, long expire) {
name = name + "_lock";
String token = UUID.randomUUID().toString();
RedisConnectionFactory factory = stringRedisTemplate.getConnectionFactory();
RedisConnection conn = factory.getConnection();
try {
//参考redis命令：
//set key value [EX seconds] [PX milliseconds] [NX|XX]
Boolean result = conn.set(
name.getBytes(),
token.getBytes(),
Expiration.from(expire, TimeUnit.MILLISECONDS),
RedisStringCommands.SetOption.SET_IF_ABSENT //NX
);
if (result != null && result)
return token;
} finally {
RedisConnectionUtils.releaseConnection(conn, factory,false);
}
return null;
}
kafka
kafka介绍
producer:发布消息的对象称之为主题生产者
topic:kafka将消息分门别类，每一类的消息称之为一个主题
consumer:订阅消息并处理发布的信息的对象称之为主题消费者
broker:已发布的信息保存在一组服务器中，称之为kafka集群，集群中的每一个服务器都是一个代理（Broker），消费者可以订阅一个或多个主题（topic）,并从Broker拉数据，从而消费这些已发布的信息。
下载镜像：docker pull zookeeper:3.4.14
创建容器
docker run -d --name zookeeper -p 2181:2181 zookeeper:3.4.14
下载镜像：
docker pull wurstmeister/kafka:2.12-2.3.1
创建容器
docker run -d --name kafka \
--env KAFKA_ADVERTISED_HOST_NAME=192.168.74.128 \
--env KAFKA_ZOOKEEPER_CONNECT=192.168.74.128:2181 \
--env KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://192.168.74.128:9092 \
--env KAFKA_LISTENERS=PLAINTEXT://0.0.0.0:9092 \
--env KAFKA_HEAP_OPTS="-Xmx256M -Xms256M" \
--net=host wurstmeister/kafka:2.12-2.3.1依赖
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka-clients</artifactId>
</dependency>生产者
package com.heima.kafka.sample;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import java.util.Properties;
/**
* 生产者
*/
public class ProducerQuickStart {
public static void main(String[] args) {
//1.kafka的配置信息
Properties properties = new Properties();
//kafka的连接地址
properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"192.168.200.130:9092");
//发送失败，失败的重试次数
properties.put(ProducerConfig.RETRIES_CONFIG,5);
//消息key的序列化器
properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringSerializer");
//消息value的序列化器
properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringSerializer");
//2.生产者对象
KafkaProducer<String,String> producer = new KafkaProducer<String, String>(properties);
//封装发送的消息
ProducerRecord<String,String> record = new ProducerRecord<String, String>("itheima-topic","100001","hello kafka");
//3.发送消息
producer.send(record);
//4.关闭消息通道，必须关闭，否则消息发送不成功
producer.close();
}
}消费者
package com.heima.kafka.sample;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import java.time.Duration;
import java.util.Collections;
import java.util.Properties;
/**
* 消费者
*/
public class ConsumerQuickStart {
public static void main(String[] args) {
//1.添加kafka的配置信息
Properties properties = new Properties();
//kafka的连接地址
properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.200.130:9092");
//消费者组
properties.put(ConsumerConfig.GROUP_ID_CONFIG, "group2");
//消息的反序列化器
properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
//2.消费者对象
KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(properties);
//3.订阅主题
consumer.subscribe(Collections.singletonList("itheima-topic"));
//当前线程一直处于监听状态
while (true) {
//4.获取消息
ConsumerRecords<String, String> consumerRecords = consumer.poll(Duration.ofMillis(1000));
for (ConsumerRecord<String, String> consumerRecord : consumerRecords) {
System.out.println(consumerRecord.key());
System.out.println(consumerRecord.value());
}
}
}
}SpringBoot集成kafka
依赖
<dependencies>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>
<!-- kafkfa -->
<dependency>
<groupId>org.springframework.kafka</groupId>
<artifactId>spring-kafka</artifactId>
<exclusions>
<exclusion>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka-clients</artifactId>
</exclusion>
</exclusions>
</dependency>
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka-clients</artifactId>
</dependency>
<dependency>
<groupId>com.alibaba</groupId>
<artifactId>fastjson</artifactId>
</dependency>
</dependencies>yml配置
server:
port: 9991
spring:
application:
name: kafka-demo
kafka:
bootstrap-servers: 192.168.200.130:9092
producer:
retries: 10
key-serializer: org.apache.kafka.common.serialization.StringSerializer
value-serializer: org.apache.kafka.common.serialization.StringSerializer
consumer:
group-id: ${spring.application.name}-test
key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
value-deserializer: org.apache.kafka.common.serialization.StringDeserializer消息生产者
package com.heima.kafka.controller;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;
@RestController
public class HelloController {
@Autowired
private KafkaTemplate<String,String> kafkaTemplate;
@GetMapping("/hello")
public String hello(){
kafkaTemplate.send("itcast-topic","黑马程序员");
return "ok";
}
}消息消费者
package com.heima.kafka.listener;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;
import org.springframework.util.StringUtils;
@Component
public class HelloListener {
@KafkaListener(topics = "itcast-topic")
public void onMessage(String message){
if(!StringUtils.isEmpty(message)){
System.out.println(message);
}
}
}传递消息为对象
发送消息
@GetMapping("/hello")
public String hello(){
User user = new User();
user.setUsername("xiaowang");
user.setAge(18);
kafkaTemplate.send("user-topic", JSON.toJSONString(user));
return "ok";
}接收消息
package com.heima.kafka.listener;
import com.alibaba.fastjson.JSON;
import com.heima.kafka.pojo.User;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;
import org.springframework.util.StringUtils;
@Component
public class HelloListener {
@KafkaListener(topics = "user-topic")
public void onMessage(String message){
if(!StringUtils.isEmpty(message)){
User user = JSON.parseObject(message, User.class);
System.out.println(user);
}
}
}MongDB
用户的搜索记录，需要给每一个用户都保存一份，数据量较大，要求加载速度快，通常这样的数据存储到mongodb更合适，不建议直接存储到关系型数据库中
拉取
拉取镜像：docker pull mongo
创建容器: docker run -di --name mongo-service --restart=always -p 27017:27017 -v ~/data/mongodata:/data mongo依赖
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-data-mongodb</artifactId>
</dependency>配置
server:
port: 9998
spring:
data:
mongodb:
host: 192.168.200.130
port: 27017
database: leadnews-history方法
package com.itheima.mongo.test;
import com.itheima.mongo.MongoApplication;
import com.itheima.mongo.pojo.ApAssociateWords;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.data.domain.Sort;
import org.springframework.data.mongodb.core.MongoTemplate;
import org.springframework.data.mongodb.core.query.Criteria;
import org.springframework.data.mongodb.core.query.Query;
import org.springframework.test.context.junit4.SpringRunner;
import java.util.Date;
import java.util.List;
@SpringBootTest(classes = MongoApplication.class)
@RunWith(SpringRunner.class)
public class MongoTest {
@Autowired
private MongoTemplate mongoTemplate;
//保存
@Test
public void saveTest(){
/*for (int i = 0; i < 10; i++) {
ApAssociateWords apAssociateWords = new ApAssociateWords();
apAssociateWords.setAssociateWords("黑马头条");
apAssociateWords.setCreatedTime(new Date());
mongoTemplate.save(apAssociateWords);
}*/
ApAssociateWords apAssociateWords = new ApAssociateWords();
apAssociateWords.setAssociateWords("黑马直播");
apAssociateWords.setCreatedTime(new Date());
mongoTemplate.save(apAssociateWords);
}
//查询一个
@Test
public void saveFindOne(){
ApAssociateWords apAssociateWords = mongoTemplate.findById("60bd973eb0c1d430a71a7928", ApAssociateWords.class);
System.out.println(apAssociateWords);
}
//条件查询
@Test
public void testQuery(){
Query query = Query.query(Criteria.where("associateWords").is("黑马头条"))
.with(Sort.by(Sort.Direction.DESC,"createdTime"));
List<ApAssociateWords> apAssociateWordsList = mongoTemplate.find(query, ApAssociateWords.class);
System.out.println(apAssociateWordsList);
}
@Test
public void testDel(){
mongoTemplate.remove(Query.query(Criteria.where("associateWords").is("黑马头条")),ApAssociateWords.class);
}
}分布式任务调度
在分布式架构下，一个服务往往会部署多个实例来运行我们的业务，如果在这种分布式系统环境下运行任务调度，我们称之为分布式任务调度。
