
概述：

- kafka消费者、消费者组和消费者位移？
- kafka幂等性以及避免重复消费？
- go kafka consumer 封装

背景：和producer同样的用途，统计博客网页浏览数据

# kafka消费者、消费者组和消费者位移？

kafka consumer 从 broker 中 pull 数据，producer push 数据到 broker 中。

优点：consumer可以承受能力的范围内控制消费速度，而且可以批量获取消息块

缺点：如果broker中没有数据，consumer会有轮询broker接口，这里kafka提供了参数来阻塞fetch请求直到有数据可以消费

**消费者组的特性：**

1. 一个消费者组可以有一个或者多个消费者
2. topic的一个partition只能有一个消费者组中的一个消费者消费，要想实现分布式消费，必须要有多个patition

**消费者位移：**

kafka面临的问题，如何让broker 和 consumer 就被消费的数据保持一致性

- broker发送消息给consumer之后立即标记为consumed。如果 consumer 崩溃或者请求超时或者其他原因，就会导致消息丢失
- 增加确认机制，消息发送标记消息为sent，consumer消费成功后标记消息为consumed。
    - 如果consumer消费成功了，但是在发送确认之前崩溃了，broker需要再次发送消息，就会导致消息重复消费。
    - 性能问题：broker需要为每个消息保存多个状态，状态的变动需要加锁
    - 如果消息一直得不到确认该如何处理

kafka的是如何实现消息确认机制的：

每一个 partition 在任意给定的时间内只能被每个订阅了这个 topic 的 consumer 组中的一个 consumer 消费，只需要保存这个consumer下一条要消费的消息的位移即可

- 同一个消费者组的消费者 不会并发消费 同一个partition，不需要加锁
- 不需要保存多个状态，消费者确认要消费到了哪一条，只想下一条即可。
- 如果consumer消费逻辑出现bug，可以在 bug 修复后通过回退到之前的 offset 来再次消费这些数据 （重要）, go 可以指定出初始位移`conf.Consumer.Offsets.Initial`来再次消费



# kafka幂等性以及避免重复消费

正常情况下producer push一条消息到broker，然后consumer从broker pull消息然后消费，这只是理想状态的情况下。我们需要考虑的是发生异常的情况kafka如何做到保证消息是不丢失，不重复消费的。

- producer 发布消息的持久性保证
- consumer 消费消息的保证，只被消费一次

**如果一个 producer 在试图发送消息的时候发生了网络故障， 则不确定网络错误发生在消息提交之前还是之后，broker到底有没有收到这条消息并持久化**？

答：如果我们想要确保消息可以重传，但绝不丢失。kafka 0.11.0.0 之前的版本就是重发机制，之后的版本增加了幂等性，会为每条消息分配序列号来避免在broker log中产生重复消息。同时kafka提供了事务机制，多条消息写入log要么都成功要么都失败。

kafka所有的副本都有相同的 log 和相同的 offset。consumer 负责控制它在 log 中的位置，理想状态下consumer永远不会崩溃，没问题。**如果consumer崩溃了，这个topic partition必须要有另一个进程来接管，此时该从哪个offset开始接管？**

1. consumer先读取消息，然后将它的位置保存在log中，在处理相应的消息，如果异常发生在位置保存后，即使有其他进程接管，这条消息可能也会丢失了
2. consumer先读取消息，处理消息，最后保存它的位置，如果异常是在消息处理完成后，还没有保存消息时。消息可能会被重复处理。

答：2种解决方案，步骤就是先读取，在消费，最后更新位移，这样可以保证消息不丢失，可能会造成消息重复消费

- 简单的方案：幂等性保证，使用offset作为唯一键，数据和offset一起写库，如果offset不存在就先建，否则更新。
- 复杂的方案：2步提交，消费完成写库，状态为待提交，更新kafka log offset，更新库的状态为已提交，此方案会复杂化问题，推荐幂等性保证，简单有效。

重点：consumer消费数据时，要做一次幂等性消费，避免重复消费数据



# kafka 消费者 & 分区

Kafka 消费者从属于消费者群组。一个群组里的消费者订阅的是同一个主题，每个消费者 接收主题一部分分区的消息。

- 假设主题 T1 有 4 个分区，我们创建了消费者 C1，它是群组 G1 里唯一的消费者，我们用 它订阅主题 T1。消费者 C1 将收到主题 T1 全部 4 个分区的消息
- 如果在群组 G1 里新增一个消费者 C2，那么每个消费者将分别从两个分区接收消息。假设消费者 C1 接收分区 0 和分区 2 的消息，那么消费者 C2 节会接收分区 1 和分区 3 的消息
- 如果群组 G1 有 4 个消费者，那么每个消费者可以分配到一个分区
- 如果群组添加更多的消费者，超过主题的分区数量，那么有一部分消费者就会被 闲置，不会接收到任何消息

**再均衡**：分区的所有权从一个消费者转移到另一个消费者

消费者通过向consumer-group 协调器的broker发送心跳来维持他们和群组的从属关系以及它们对分区的所有权关系，只要消费者以正常的时间间隔发送心跳，就被认为是活跃的，说明还在正常读取分区的消息，

如果消费者发生崩溃，consumer-group协调器broker，在确认了该consumer死亡后会出发再均衡。

`max.poll.interval.ms` 控制发送心跳的间隔时间

`session.timeout.ms` 控制失联多久判定为死亡

# go kafka consumer 

封装一下go consumer支持

- 优雅关闭consumer
- 记录consumer异常，同producer一样Consumer.Return channel设置为true以后必须有相应的读取，否则会阻塞
- kafka崩溃恢复，可重新加入consumer group，恢复消费

```go
type ConsumerGroup interface {
	Consume(topic []string, handler sarama.ConsumerGroupHandler)
}

type consumerGroup struct {
	ctx     context.Context
	client  sarama.ConsumerGroup
	groupId string
}

func NewConsumerGroup(ctx context.Context, groupId string) ConsumerGroup {
	conf := sarama.NewConfig()
	conf.Version = sarama.V0_11_0_2
	conf.Consumer.Return.Errors = true
	client, err := sarama.NewConsumerGroup(config.Kafka.Broker(), groupId, conf)
	if err != nil {
		panic(errors.Wrap(err, "kafka new consumer client err"))
	}
	// Track errors
	go func() {
		for err := range client.Errors() {
			logger.Wrapf(err, "kafka %s consume err", groupId)
		}
	}()
	go func() {
		for {
			select {
			// 主进程退出，通知consumer关闭
			case <-ctx.Done():
				_ = client.Close()
				//logger.Infof("quit: kafka consumer %s", groupId)
				return
			}
		}
	}()
	return &consumerGroup{
		ctx:     ctx,
		client:  client,
		groupId: groupId,
	}
}

func (c *consumerGroup) Consume(topic []string, handler sarama.ConsumerGroupHandler) {
	go func() {
		defer c.client.Close()
		for {
			err := c.client.Consume(c.ctx, topic, handler)
			if err != nil {
				switch err {
				case sarama.ErrClosedClient, sarama.ErrClosedConsumerGroup:
					// 退出
					logger.Infof("quit: kafka consumer %s", c.groupId)
					return
				case sarama.ErrOutOfBrokers:
					logger.Errorf("kafka 崩溃了~")
				default:
					logger.Errorf("kafka exception: %s", err.Error())
				}
				time.Sleep(1 * time.Second)
			}
		}
	}()
}
```
---

go kafka consumer group 使用姿势


prometheus 是go语言编程的，可以直接下载对应系统的二进制文件运行即可。

**预编译二进制安装**

下载地址: 

- prometheus [https://prometheus.io/download/](https://prometheus.io/download/)
  - centos 下载 `.linux-amd64.tar.gz` 即可
- grafana https://grafana.com/grafana/download

prometheus安装:

```shell
tar -zxvf prometheus-2.25.0.linux-amd64.tar.gz
./prometheus
```

grafana安装：

```shell
wget https://dl.grafana.com/oss/release/grafana-7.4.2-1.x86_64.rpm
sudo yum install grafana-7.4.2-1.x86_64.rpm
```

**docker安装prometheus/grafana**

`docker-compose.md`

```yml
version: '3.2'
services:
  prometheus:
    image: prom/prometheus
    container_name: prometheus
    ports:
      - "9090:9090"
    volumes:
      - ./prometheus.yml:/etc/prometheus/prometheus.yml
    networks:
      - prometheus_network
  grafana:
    image: grafana/grafana
    container_name: grafana
    ports:
      - 3000:3000
    networks:
      - prometheus_network
networks:
  prometheus_network:
    driver: bridge
```

`prometheus.yml`

```yml
# my global config
global:
  scrape_interval:     15s # Set the scrape interval to every 15 seconds. Default is every 1 minute.
  evaluation_interval: 15s # Evaluate rules every 15 seconds. The default is every 1 minute.
  # scrape_timeout is set to the global default (10s).

# Alertmanager configuration
alerting:
  alertmanagers:
  - static_configs:
    - targets:
      # - alertmanager:9093

# Load rules once and periodically evaluate them according to the global 'evaluation_interval'.
rule_files:
  # - "first_rules.yml"
  # - "second_rules.yml"

# A scrape configuration containing exactly one endpoint to scrape:
# Here it's Prometheus itself.
scrape_configs:
  # The job name is added as a label `job=<job_name>` to any timeseries scraped from this config.
  - job_name: 'prometheus'
    # metrics_path defaults to '/metrics'
    # scheme defaults to 'http'.
    static_configs:
    - targets: ['docker.for.mac.host.internal:9003']
```

docker 调用宿主机服务，使用 `docker.for.mac.host.internal` 作为host

**grafana设置prometheus datasource**

![image-20210219173605639](/images/blog/1/prometheus/assert/prometheus-grafana-datasource.png)



Prometheus Grafana 安装配置

概述

- kafka 库
- producer config 配置
- sync vs async
- 压缩方式

背景：博客文章想统计访问数据，每日pv/uv/ip数据

分析：

- 网页浏览数据，主要以写为主
- 时效性、事务持久要求不高
- 数据量大

kafka特别适合这种场景，高吞吐量，较低的端到端延迟，强大的持久性。

# go kafka 库

- [github.com/Shopify/sarama](https://pkg.go.dev/github.com/Shopify/sarama)  使用人数较多，我公司也是用的这个库
- [github.com/confluentinc/confluent-kafka-go](https://github.com/confluentinc/confluent-kafka-go) 我没有做过实验，有兴趣的朋友自己尝试一下

```
go get github.com/Shopify/sarama
```

# sync vs async

同步异步的区别, 如果因为实时性要求没那么高的话，推荐使用async，总结一下原因：kafka面临的2大难题之一，**持久化要面临磁盘大量的I/O操作**，如此频繁的写，即便是顺序写、有pagecache可以缓冲、有mmap减少一次内核态的复制，磁盘也是难以承受。

kafka的处理也是很简单，加缓存，本来1毫秒100次写，broker缓存10次一写（数据块），这样可以减轻10倍的I/O压力。

所以这里也是同样的思想，kafka加了一层缓存，我们producer也可以在加一层缓存，又可以减轻n倍的I/O压力。所以在实时性要求不高的前提下，推荐使用async

async有4个配套参数

```go
conf := sarama.NewConfig()
conf.Producer.Flush.Messages = 10                      // 缓存条数
conf.Producer.Flush.Frequency = 500 * time.Millisecond // 缓存时间

Flush struct {
  // 达到多少字节时，触发一次broker请求
  Bytes int
  // 达到多少条消息时，触发一次broker请求
  Messages int 
  // producer缓存消息的时间, 默认缓存500毫秒 => queue.buffering.max.ms
  Frequency time.Duration 
  // producer缓存队列里最大缓存的消息数量 => queue.buffering.max.messages
  MaxMessages int
}
```

完整的prodcuer用法

```go
type Produce struct {
	AsyncProducer sarama.AsyncProducer
}

func NewProducer(ctx context.Context) *Produce {
	conf := sarama.NewConfig()
	conf.Producer.RequiredAcks = sarama.WaitForAll
	conf.Producer.Return.Errors = true                     // 接收producer的error，下面细说用法
	conf.Producer.Compression = sarama.CompressionLZ4      // 压缩方式，如果kafka版本大于1.2，推荐使用zstd压缩
	conf.Producer.Flush.Messages = 10                      // 缓存条数
	conf.Producer.Flush.Frequency = 500 * time.Millisecond // 缓存时间

	producer, err := sarama.NewAsyncProducer(config.Kafka.Broker(), conf)
	if err != nil {
		panic(errors.Wrapf(err, "failed to open kafka producer: "))
	}
	pd := &Produce{AsyncProducer: producer}
	// 接受结束信号
	go func() {
		for {
			select {
			case err := <-producer.Errors():
				logger.Errorf("kafka producer send error %s", err.Err.Error())
			case <-ctx.Done(): // 接收主进程结束信号，持久化缓存数据，防止丢失，关闭producer连接
				_ = producer.Close()
				logger.Infof("quit: kafka producer")
				return
			}
		}
	}()
	return pd
}

func (p *Produce) Send(topic string, value []byte) {
	p.AsyncProducer.Input() <- &sarama.ProducerMessage{
		Topic: topic,
		Value: sarama.ByteEncoder(value),
	}
}
```

- `conf.Producer.Return`: 这里是go语言的专属用法，有2个channel，**重点：**如果 `conf.Producer.Return.Errors = true` 设置为了true，就必须有相应读取channel，否则会阻塞producer写消息，success channel/error channel默认capacity只有256字节 <a name="producer_return"></a>
    - `success`: 写消息成功channel 
    -  `errors`:  错误channel
- `conf.Producer.RequiredAcks` 对应java producer的 `request.required.acks`
    - `sarama.NoResponse` - `0`:   producer不等待来自broker同步完成的确认继续发送下一条（批）消息。延迟最低/消息持久化保证性最低（当服务器发生故障时某些数据会丢失，如leader已死，但producer并不知情，发出去的信息broker就收不到）。
    - `sarama.WaitForLocal` - `1`: producer在leader已成功收到的数据并得到确认后发送下一条message，没有确认其他副本是否收到。持久化保证较好（被写入死亡leader但尚未复制将失去了唯一的消息）
    - `sarama.WaitForAll` - `2`（默认）:  producer在leader以及follower副本确认接收到数据后才算一次发送完成。 

- `conf.Producer.Compression` : kafka 消息压缩了，综合考量来说，如果kafka版本大于2.1可以使用zstd，小于2.1版本可以使用lz4, 无论是cpu占用还是压缩比例的角度上看

| 算法           | cpu  | 压缩比（带宽同理） | producer吞吐量(压缩快慢) | consumer吞吐量(解压缩快慢) |
| -------------- | ---- | ------------------ | ------------------------ | -------------------------- |
| Gzip           | 1    | 3                  | 4                        | 3                          |
| Snappy         | 4    | 4                  | 2                        | 2                          |
| LZ4            | 3    | 2                  | 1                        | 1                          |
| ZSTD (2.1版本) | 2    | 1                  | 3                        | 4                          |

# issue

1. go kafka producer在写消息时，写了1，2条就阻塞了？ 

答：[conf.Producer.Return配置姿势不对？](#producer_return) 

`conf.Producer.Return`: 这里是go语言的专属用法，有2个channel，**重点：**如果 `conf.Producer.Return.Errors = true` 设置为了true，就必须有相应读取channel，否则会阻塞producer写消息，success channel/error channel默认capacity只有256字节 <a name="producer_return"></a>

- `success`: 写消息成功channel 
-  `errors`:  错误channel



go kafka producer在写消息时，写了1，2条就阻塞了？

go kafka producer 实践与落地


数据量大的话：mysql是通过partition分区表支持按日期来分区，es通过index名称来达到分区的策略

es可以通过put mapping来创建，但是每天都要新的index，mysql需要写脚本自动来添加partition。es提供了template来完成，当POST doc时，如果不存在会匹配template初始化mapping

**创建template**

支持通配符匹配，如果匹配到多个，按照order优先排序使用

```
PUT _template/tracking_template
{
  "order": 0,
  "index_patterns": [
    "spider_*"
  ],
  "settings": {},
  "mappings": {
    "properties": {
      "date": {
        "type": "date"
      },
      "spider": {
        "type": "keyword"
      }
    }
  },
  "aliases": {}
}
```

**查看索引模版**

```curl
GET _template                							// 查看所有模板
GET _template/spider_*          					// 查看与通配符相匹配的模板
GET _template/spider_2020,spider_2021   	// 查看多个模板
GET _template/spider_template  						// 查看指定模板
```

**删除索引模版**

```curl
DELETE _template/spider_template
```

使用es存储log日志来分析数据，需要按日期来分割数据（douyacun_log-2021-02）定时删除历史数据

es使用template自动初始化mapping


## react天气组件

 [![npm](https://img.shields.io/npm/v/react-tencent-weather)](https://www.npmjs.com/package/react-tencent-weather) [![npm](https://img.shields.io/npm/dm/react-tencent-weather)](https://www.npmjs.com/package/react-tencent-weather) [![npm](https://img.shields.io/npm/l/react-tencent-weather)](https://www.npmjs.com/package/react-tencent-weather) ![echarts supported](https://img.shields.io/badge/echarts-%5E3.0.0%20%7C%7C%20%5E4.0.0-blue.svg) ![react supported](https://img.shields.io/badge/React-%5E16.8.0-blue.svg)

体验地址: [https://www.douyacun.com](https://www.douyacun.com)

文档：[https://www.douyacun.com/article/b9093956c565e471f45c035b5870ed0c](https://www.douyacun.com/article/b9093956c565e471f45c035b5870ed0c)

特点：

1. 使用腾讯天气接口: `https://wis.qq.com/weather/common?source=pc&weather_type=forecast_1h|forecast_24h|alarm|limit|tips|rise|observe|index&province=${province}&city=${city}&county=`
2. 支持：逐小时预报/7日天气预报/生活指数
3. 支持IP定位，使用[免费IP定位接口](https://www.douyacun.com/article/a57b58a343f051cf1fb9761a31d37693)
4. 支持服务端渲染

![react腾讯天气组件](/images/blog/1/react/assert/image-20210107225932253.png)
![react腾讯天气组件](/images/blog/1/react/assert/image-20210107230116155.png)


## install

```bash
npm install react_tencent_weather --save
```

运行demo:

```bash
git clone https://github.com/douyacun/react-tencent-weather.git

npm install

npm start
```

本地浏览器打开 [http://127.0.0.1:3000](http://127.0.0.1:3000)

## usage

客户端使用

```react
import React from 'react';
import ReactDOM from 'react-dom';
import Weather from 'react-tencent-weather';

ReactDOM.render(
    <div><Weather province="上海" city="上海" /></div>,
    document.getElementById('root')
)
```

服务端渲染

```react
import React from 'react';
import ReactDOM from 'react-dom';
import Weather from 'react-tencent-weather/lib/ssr/index.js';
import 'react-tencent-weather/lib/ssr/index.css';

ReactDOM.render(
    <div><Weather province="上海" city="上海" /></div>,
    document.getElementById('root')
)
```

## component props

- `province`: (optional, string)

省份，默认上海

- `city`: (optional, string)

城市，默认上海

- `showDays` : (optional, bool)

是否显示 7日天气预报

- `showHours`: (optional, bool)

是否显示 逐小时预报

- `showLiving`: (optional, bool)

是否显示 生活指数

- `showTomorrow`:  (optional, bool)

是否显示明日预报

# LICENSE

MIT@[douyacun](https://github.com/douyacun)

关注我的微信公众号

![douyacun](/images/blog/1/react/assert/douyacun_qrcode.jpg)


博客右边框空荡荡的，需要一个天气组件来支撑一下，就写了这个天气组件～

react天气组件: 腾讯天气


**敬请周知**：

1. 添加友情链接的2种方式
    1. **注册并且在此文章底部评论以后** 会自动添加（推荐） :
        1. 恁叫啥 -> 站点名称 eg: Douyacun
        2. 站点 -> 博客的地址 eg: https://www.douyacun.com
    2. 发个邮件到 douyacun@gmail.com ，写明站点名称/站点连接
2. 小技巧：
    1. `command + alt + u` 查看源码必须包含有站点名称和站点链接
    2. `<a>` 标签里面不要 `alt="nofollow"`
    3. 确保你的站点已经添加了友情链接 `<a href="https://www.douyacun.com" target="_blank">Douyacun</a>` 
    4. `robots.txt` 不要 `Disallow`
3. 目前仅有20个坑位

![douyacun入驻](/images/blog/1/敬请周知/assert/douyacun入驻.png)



敬请周知：如何在Douyacun添加友情链接?


# yapi 设置环境变量，如何修改全局变量？

`环境配置` 一项可以添加该项目下接口的实际环境，供 [接口测试](https://hellosean1025.github.io/yapi/documents/case.html) 使用，这里增加了全局 header，可以在项目中设置全局 header 值

![截屏2020-12-19 下午2.03.31](/images/blog/1/tools/assert/yapi-环境变量配置.png)

可以设置全局的header，cookie，global，使用 `{{}}`来引用环境变量，但是大部分使用环境变量的场景是需要能自动修改环境变量，比如：调用登录接口获取token，其他接口在调用时可以自动设置 `Authorization: Bearer helloworld`这种场景，如果使用global，yapi没有提供可以修改global的接口，做不到



那怎么办？我肯定不会每次token过期了再去手动设置global access_token的！



思路：

yapi提供了 storage 来存储数据 和  pre-script/pre-script 请求前后处理脚本。

每次调用登录接口后存储 access_token到storage中, 然后每次请求拼接设置header  `Authorization: Bearer helloworld`

`Pre-request Script(请求参数处理脚本)`

```js
context.requestHeader.Authorization = "Bearer " + storage.getItem('accessToken');
```

`Pre-response Script(响应数据处理脚本)`

```js
if (context.responseData && context.responseData.data && context.responseData.data.accessToken) {
    storage.setItem("accessToken", context.responseData.data.accessToken)
}

console.log("accessToken", context.responseData.data.accessToken);
```


![](/images/blog/1/tools/assert/yapi-pre-script.png)

注意必须保存才能生效的，不生效看看有没有保存成功，字段是否一致？

请求成功案例：

[YApi 新版如何查看 http 请求数据](https://juejin.cn/post/6844903795743260685)

![yapi-环境变量-成功案例.png](/images/blog/1/tools/assert/yapi-环境变量-成功案例.png)

yapi无法修改全局变量！如何曲线救国！

yapi 设置环境变量，如何修改全局变量？


**ES初始化密码**

```shell
./elasticsearch-setup-passwords interactive
```

**ES开启密码校验**

```ini
xpack.security.enabled: true
xpack.security.transport.ssl.enabled: true
```

**ES(7.*)集群配置**

```ini
path.data: /data/elasticsearch/data
path.logs: /data/elasticsearch/logs

# 集群名称，节点集群名称必须一样
cluster.name: douyacun
# 节点id，唯一
node.name: node-2
# 限制客户端IP地址
network.bind_host: 0.0.0.0
# 集群节点连接IP
network.publish_host: 116.196.69.34
# http端口
http.port: 9200
# 集群节点之间 TCP 端口
transport.port: 9300
# 是否参加master节点选举
node.master: true
# 是否允许节点写数据
node.data: true
# 集群其他节点
discovery.seed_hosts: ["116.196.69.34:9300"]
# 允许选举master的节点
cluster.initial_master_nodes: ["node-2"]
```

**ES集群密码，证书配置**

生成证书

```shell
./bin/elasticsearch-certutil ca
./bin/elasticsearch-certutil cert --ca elastic-stack-ca.p12
```

配置，`elastic-certificates.p12` 必须放在`config`目录下面

```ini
xpack.security.enabled: true
xpack.security.transport.ssl.enabled: true
xpack.security.transport.ssl.verification_mode: certificate
xpack.security.transport.ssl.keystore.path: elastic-certificates.p12
xpack.security.transport.ssl.truststore.path: elastic-certificates.p12
```

https证书，通过nginx代理服务器配置就可以了，不建议es配置https证书

**Kibana设置ES账户密码**

```ini
elasticsearch.username: "kibana"
elasticsearch.password: "密码"
```

**完整配置**

```ini
cluster.name: douyacun
path.data: /data/elasticsearch/data
path.logs: /data/elasticsearch/logs

bootstrap.memory_lock: true
network.bind_host: 0.0.0.0
network.publish_host: 116.196.69.34
http.port: 9200
transport.port: 9300
node.name: node-2
node.master: true
node.data: true
discovery.seed_hosts: ["116.196.69.34:9300"]
cluster.initial_master_nodes: ["node-2"]

xpack.security.enabled: true
xpack.security.transport.ssl.enabled: true
xpack.security.transport.ssl.verification_mode: certificate
xpack.security.transport.ssl.keystore.path: elastic-certificates.p12
xpack.security.transport.ssl.truststore.path: elastic-certificates.p12
```



es集群,证书命令


背景：之前开发的聊天是的功能，用的人太少了，考虑再三就决定去掉聊天室的功能了，改做一些其他有意思的事情，比如目前已经开发好了的左边栏当前在线人数，阅读时长统计（正在开发中）

遇到的问题：

- 页面虽然是react写的，但是nextjs服务端渲染的，不是单页应用就不能像单页应用那样有全局变量存储整站的websocket
- 服务端使用nginx作为代理服务器（之前是kong，后来干掉了），nginx proxy模块默认有 read_timeout/send_timeout 默认值 60s
- 服务端socket存储数据结构选型
- 安全防护

实现思路：

因为是多页模式，就需要每个页面初始化一个wss连接，按cookie去重以后就是当前在线人数。

页面wss重联机制，毕竟断网是很常见的

```jsx
/**
* websocket: 初始化 
*/
const connect = () => {
  conn = new WebSocket(ws_address);
  conn.onmessage = handlerMessage;
  // ws 关闭需要重新连接，每隔1s尝试1次
  conn.onclose = function (e) {
    console.log('Socket is closed. Reconnect will be attempted in 1 second.', e.reason);
    setTimeout(function () {
      connect();
    }, 1000);
  };
  // 连接失败
  conn.onerror = function (e) {
    console.error('Socket encountered error: ', e.message, 'Closing socket');
    conn.close()
  }
}
```



nginx：proxy_read_timeout/proxy_send_timeout默认60s超时，心跳检测，只需要服务端在60s内定时发送ping(0x9)消息就可以了，浏览器会自动回复pong消息

```go
for {
  select {
    // 定时向客户端推送，当前在线人数
    case <-countTicker.C:
    c.send <- c.hub.Count().Bytes()
    // 定时心跳检测
    case <-pingTicker.C:
    _ = c.conn.SetWriteDeadline(time.Now().Add(writeWait))
    if err := c.conn.WriteMessage(websocket.PingMessage, nil); err != nil {
      return
    }
  }
}
```



服务端存储socket结构体，没有向指定socket写数据的需求，不需要额外维护socket的数据结构，维护一个cookie在线页面数即可

```go
type Hub struct {
	// Registered uid
	uuid map[string]int
	// Inbound messages from the clients.
	broadcast chan Responser
	// Register requests from the clients.
	register chan *Client
	// Unregister requests from clients.
	unregister chan *Client
}

func (h *Hub) Run() {
  for {
    select {
      case client := <-h.register:
      // 注册ws
      h.uuid[client.uuid]++
      client.send <- hub.Count().Bytes()
      case client := <-h.unregister:
      // 删除ws
      h.uuid[client.uuid]--
      if h.uuid[client.uuid] == 0 {
        delete(h.uuid, client.uuid)
      }
    }
  }
```

安全方面需要考虑的内容是：

1. 如何检测跨站点 WebSocket 劫持漏洞?
2. 同一设备  ws 连接数如何限制，避免bug导致连接数过多导致服务挂掉?

解决方法：

1. check origin, 确保是自己网站的来源
2. 服务器端为每个 WebSocket 客户端生成唯一的一次性 Token；
3. 客户端将 Token 作为 WebSocket 连接 URL 的参数（譬如 wss://www.douyacun.com/?token＝randomOneTimeToken），发送到服务器端进行 WebSocket 握手连接；
4. 服务器端验证 Token 是否正确，一旦正确则将这个 Token 标示为废弃不再重用，同时确认 WebSocket 握手连接成功；如果 Token 验证失败或者身份认证失败，则返回 403 错误。
5. 限流，限制设备最多最大链接数，必要时，deny ip地址

鉴于之前开发的聊天功能没人用，考虑在三去掉了聊天功能，改成统计当前在线人数

websocket如何实现统计当前在线人数



# gositemap

Sitemap 协议: https://www.sitemaps.org/protocol.html

博客: [https://www.douyacun.com](https://www.douyacun.com)

博客 sitemap: [https://www.douyacun.com/sitemap.xml](https://www.douyacun.com/sitemap.xml)

Github: [https://github.com/douyacun/gositemap](https://github.com/douyacun/gositemap)

pkg.go.dev: [https://pkg.go.dev/github.com/douyacun/gositemap](https://pkg.go.dev/github.com/douyacun/gositemap)

### Installation

```
import "github.com/douyacun/gositemap"
```

go语言实现的sitemap生成工具
```go
st := gositemap.NewSiteMap()
st.SetPretty(true)

url := gositemap.NewUrl()
url.SetLoc("https://www.douyacun.com/")
url.SetLastmod(time.Now())
url.SetChangefreq(Daily)
url.SetPriority(1)
st.AppendUrl(url)
bt, err := st.ToXml()
if err != nil {
  fmt.Printf("%v", err)
  return
}
fmt.Printf("%s", bt)
```
输出
```xml
<?xml version="1.0" encoding="UTF-8"?>
<urlset xmlns="http://www.sitemaps.org/schemas/sitemap/0.9">
  <url>
    <loc>https://www.douyacun.com/</loc>
    <lastmod>2020-04-19T17:28:33+08:00</lastmod>
    <changefreq>daily</changefreq>
    <priority>1</priority>
  </url>
</urlset>
```

### Features

- [x]  [gositemap](#gositemap)
- [x]  [Image sitemap](#image-sitemap)
- [x]  [News sitemap](#news-sitemap)
- [x]  [Video sitemap](#video-sitemap)
- [x]  [file storage](#file-storage)
- [x]  [Sitemap index](#sitemap-index)

### Image sitemap

Google 图片扩展功能 [Google Image Support: 
https://support.google.com/webmasters/answer/178636?hl=zh-Hans&ref_topic=4581190

- **提供适当的相关信息**：确保您的视觉内容与其所在网页的主题相关。我们建议您仅在能为网页增添原创价值的情况下展示图片。我们极不赞成在网页中完全使用非原创的图片和文字内容。
- **优化放置位置**：尽可能将图片放置在相关文字附近。必要时，也可考虑将最重要的图片放置在网页顶部附近。
- **勿将重要文字内嵌在图片中**：避免将文字（特别是网页标题和菜单项等重要的文字元素）内嵌在图片中，因为并非所有用户都能访问这类文字（而且网页翻译工具不适用于图片）。为了尽可能让更多的人能访问您的内容，请使用 HTML 格式提供文本，并为图片提供替代文本。
- **创建信息丰富的优质网站**：对 Google 图片而言，优质的网页内容与视觉内容同等重要 - 它可以提供背景信息并更能吸引用户点击搜索结果。网页内容可用于为图片生成一段文本摘要，而且 Google 在进行图片排名时会考虑对应的网页内容质量
- **创建适合在各种设备上访问的网站**：比起桌面设备，用户更多地使用移动设备在 Google 图片上进行搜索。因此，有必要设计一个适合所有设备类型和尺寸的网站。请使用移动设备适合性测试工具测试您的网页在移动设备上的运行效果，并获取反馈以了解哪些内容需要修正。
- **为图片创建良好的网址结构**：Google 会借助网址路径以及文件名来理解您的图片。我们建议您好好组织图片内容，以使网址结构合乎逻辑。

```go
st := NewSiteMap()
st.SetPretty(true)
url := NewUrl()
url.SetLoc("https://www.douyacun.com/show_image")
// 注意这里url.SetLoc设置网页的网址
// image.SetLoc 设置的是图片访问路径，image的域名和网址域名不一致也可以
_image := NewImage().
    SetLoc("https://www.douyacun.com/image1.jpg").
    SetTitle("example").
    SetCaption("图片的说明。").
    SetLicense("https://www.douyacun.com")
url.AppendImage(_image)

_image = NewImage().
    SetLoc("https://www.douyacun.com/image2.jpg").
    SetTitle("example").
    SetCaption("图片的说明。").
    SetLicense("https://www.douyacun.com")
url.AppendImage(_image)

st.AppendUrl(url)
data, err := st.ToXml()
if err != nil {
    fmt.Printf("%v", err)
    return
}
fmt.Printf("%s", data)
```

输出
```xml
<?xml version="1.0" encoding="UTF-8"?>
<urlset xmlns="http://www.sitemaps.org/schemas/sitemap/0.9" xmlns:image="http://www.google.com/schemas/sitemap-image/1.1">
  <url>
    <loc>https://www.douyacun.com/show_image</loc>
    <image:image>
      <image:loc>https://www.douyacun.com/image1.jpg</image:loc>
      <image:caption>图片的说明。</image:caption>
      <image:title>example</image:title>
      <image:license>https://www.douyacun.com</image:license>
    </image:image>
    <image:image>
      <image:loc>https://www.douyacun.com/image2.jpg</image:loc>
      <image:caption>图片的说明。</image:caption>
      <image:title>example</image:title>
      <image:license>https://www.douyacun.com</image:license>
    </image:image>
  </url>
</urlset>
```

### News sitemap

Google 新闻站点地图准则 google news support
https://support.google.com/webmasters/answer/178636?hl=zh-Hans&ref_topic=4581190

```go
st := NewSiteMap()
st.SetPretty(true)

url := NewUrl()
url.SetLoc("https://www.douyacun.com/business/article55.html")
url.AppendNews(NewNews().SetName("《示例时报》").
    SetTitle("公司 A 和 B 正在进行合并谈判").
    SetLanguage("zh-cn").
    SetPublicationDate(time.Now()))
st.AppendUrl(url)
data, err := st.ToXml()
if err != nil {
    fmt.Printf("%v", err)
    return
}
fmt.Printf("%s", data)
```
输出
```xml
<?xml version="1.0" encoding="UTF-8"?>
<urlset xmlns="http://www.sitemaps.org/schemas/sitemap/0.9" xmlns:news="http://www.google.com/schemas/sitemap-news/0.9">
  <url>
    <loc>https://www.douyacun.com/business/article55.html</loc>
    <news:news>
      <news:publication>
        <news:name>《示例时报》</news:name>
        <news:language>zh-cn</news:language>
      </news:publication>
      <news:publication_date>2020-04-19T17:24:49+08:00</news:publication_date>
      <news:title>公司 A 和 B 正在进行合并谈判</news:title>
    </news:news>
  </url>
</urlset>
```

### Video sitemap

视频 Sitemap 及其替代方案  Google Video Support 
https://support.google.com/webmasters/answer/80471?hl=zh-Hans&ref_topic=4581190

- **Google 必须能够找到该视频。** 系统会根据是否存在某种 HTML 标记（例如 ``、`` 或 ``）来识别网页中的视频。请确保相应网页不[需要复杂的用户操作或特定的网址片段即可加载](https://support.google.com/webmasters/answer/156442#complex-javascript)，否则 Google 可能找不到它。**提示**：虽然我们可以通过自然抓取找到网页中内嵌的视频，但您也可以通过发布[视频 Sitemap](https://support.google.com/webmasters/answer/80471) 帮助我们找到您的视频。
- **您必须为视频提供高品质的缩略图 。**
- **确保每个视频都位于可公开访问的网页中**，用户可以在其中观看视频。该网页不应该要求用户登录，也不应该被 robots.txt 或 noindex屏蔽（必须可供 Google 访问）。
- **视频内容应确切吻合其托管网页的内容。** 例如，如果您拥有一个介绍桃饼的食谱网页，不要嵌入笼统介绍甜点的视频。
- 确保您在视频 Sitemap 或视频标记中提供的任何信息与实际视频内容**一致**。

```go
st := NewSiteMap()
url := NewUrl()
url.Loc = "https://www.douyacun.com"

video := NewVideo().
    SetThumbnailLoc("http://www.example.com/thumbs/123.jpg").
    SetTitle("适合夏季的烧烤排餐").
    SetDescription("小安教您如何每次都能烤出美味牛排").
    SetContentLoc("http://streamserver.example.com/video123.mp4").
    SetPlayerLoc("http://www.example.com/videoplayer.php?video=123", true).
    SetDuration(600*time.Second).
    SetExpirationDate(time.Now().Add(time.Hour*24)).
    SetRating(4.2).
    SetViewCount(12345).
    SetPublicationDate(time.Now().Add(-time.Hour*24)).
    SetFamilyFriendly(true).
    SetRestriction([]string{"IE", "GB", "US", "CN", "CA"}, true).
    SetPrice(6.99, "EUR", true, true).
    SetRequiresSubscription(true).
    SetUploader("GrillyMcGrillerson", "http://www.example.com/users/grillymcgrillerson").
    SetLive(true)

url.AppendVideo(video)
st.AppendUrl(url)
bt, err := st.ToXml()
if err != nil {
    log.Printf("%v", err)
    return
}
fmt.Printf("%s", bt)
```

输出
```xml
<?xml version="1.0" encoding="UTF-8"?>
<urlset xmlns="http://www.sitemaps.org/schemas/sitemap/0.9" xmlns:video="http://www.google.com/schemas/sitemap-video/1.1">
  <url>
    <loc>https://www.douyacun.com</loc>
    <video:video>
      <video:thumbnail_loc>http://www.example.com/thumbs/123.jpg</video:thumbnail_loc>
      <video:title>适合夏季的烧烤排餐</video:title>
      <video:description>小安教您如何每次都能烤出美味牛排</video:description>
      <video:content_loc>http://streamserver.example.com/video123.mp4</video:content_loc>
      <video:player_loc allow_embed="yes">http://www.example.com/videoplayer.php?video=123</video:player_loc>
      <video:duration>600</video:duration>
      <video:expiration_date>2020-04-20T17:25:49+08:00</video:expiration_date>
      <video:rating>4.2</video:rating>
      <video:view_count>12345</video:view_count>
      <video:publication_date>2020-04-18T17:25:49+08:00</video:publication_date>
      <video:family_friendly>yes</video:family_friendly>
      <video:restriction relationship="allow">IE GB US CN CA</video:restriction>
      <video:price currency="EUR" type="own" resolution="hd">6.99</video:price>
      <video:requires_subscription>yes</video:requires_subscription>
      <video:uploader info="http://www.example.com/users/grillymcgrillerson">GrillyMcGrillerson</video:uploader>
      <video:live>yes</video:live>
    </video:video>
  </url>
</urlset>
```

### file storage

```go
st := NewSiteMap()
// 默认域名
st.SetDefaultHost("https://www.douyacun.com")
st.SetPretty(true)
// 每个sitemap文件不能多于50000个链接，这里可以自己配置每个文件最多，如果超过MaxLinks，会自动生成sitemap_index.xml文件
// st.SetMaxLinks(2)
st.SetPublicPath("/tmp/gositemap")

url := NewUrl()
url.SetLoc("https://www.douyacun.com/business/article55.html")
url.AppendNews(NewNews().SetName("《示例时报》").
               SetTitle("公司 A 和 B 正在进行合并谈判").
               SetLanguage("zh-cn").
               SetPublicationDate(time.Now()))
st.AppendUrl(url)

path, err := st.Storage()
if err != nil {
  fmt.Printf("%v", err)
  return
}
fmt.Println(path)
```

生成`sitemap.xml`文件

```xml
<?xml version="1.0" encoding="UTF-8"?>
<urlset xmlns="http://www.sitemaps.org/schemas/sitemap/0.9" xmlns:news="http://www.google.com/schemas/sitemap-news/0.9">
  <url>
    <loc>https://www.douyacun.com/business/article55.html</loc>
    <news:news>
      <news:publication>
        <news:name>《示例时报》</news:name>
        <news:language>zh-cn</news:language>
      </news:publication>
      <news:publication_date>2020-04-19T17:44:33+08:00</news:publication_date>
      <news:title>公司 A 和 B 正在进行合并谈判</news:title>
    </news:news>
  </url>
</urlset>
```

### Sitemap index 

拆分较大的站点地图

```go
mapIndex := NewSiteMapIndex()

st1 := NewSiteMap()
st1.SetPretty(true)
st1.SetFilename("sitemap1")
st1.SetCompress(true)

# 以news为例
url := NewUrl()
url.SetLoc("https://www.douyacun.com/business/article55.html")
url.AppendNews(NewNews().SetName("《示例时报》").
               SetTitle("公司 A 和 B 正在进行合并谈判").
               SetLanguage("zh-cn").
               SetPublicationDate(time.Now()))
st1.AppendUrl(url)

url2 := NewUrl().SetLoc("https://www.douyacun.com/business/article56.html")
url2.AppendNews(NewNews().SetName("《示例时报》").
               SetTitle("公司 A 和 C 正在进行合并谈判").
               SetLanguage("zh-cn").
               SetPublicationDate(time.Now()))
st1.AppendUrl(url2)
st1Filename, err := st1.Storage()
if err != nil {
  fmt.Printf("%v", err)
  return
}
mapIndex.Append("https://www.douyacun.com/" + st1Filename)
filename, err := mapIndex.Storage("/Users/liuning/Documents/github/gositemap/sitemap_index.xml")
if err != nil {
  fmt.Printf("%v", err)
  return
}
fmt.Printf("%v", filename)
```

使用sitemap_index时，建议每个单独的sietmap comporess压缩成.gz文件，`SetCompress`后会自动添加 `.gz`后缀名 ,  生成`sitemap1.xml.gz` 和 `sitemap_index.xml`

### Documentation

Api Reference : https://pkg.go.dev/github.com/douyacun/gositemap?tab=doc



# LICENSE

MIT@[douyacun](https://github.com/douyacun).

![douyacun](/images/blog/1/assert/douyacun_qrcode.jpg)