日期: 2021 年 7 月 6 日

306. 累加数（JS实现）

306. 累加数（JS实现）
1 题目
累加数是一个字符串，组成它的数字可以形成累加序列。
一个有效的累加序列必须至少包含 3 个数。除了*开始的两个数以外，字符串中的其他数都等于它之前两个数相加的和。
给定一个只包含数字 ‘0’-‘9’ 的字符串，编写一个算法来判断给定输入是否是累加数。
说明: 累加序列里的数不会以 0 开头，所以不会出现 1, 2, 03 或者 1, 02, 3 的情况。
示例 1:
输入: “112358”
输出: true
解释: 累加序列为: 1, 1, 2, 3, 5, 8 。1 + 1 = 2, 1 + 2 = 3, 2 + 3 = 5, 3 + 5 = 8
示例 2:
输入: “199100199”
输出: true
解释: 累加序列为: 1, 99, 100, 199。1 + 99 = 100, 99 + 100 = 199
链接：https://leetcode-cn.com/problems/additive-number
2 思路
这道题用递归的方法，依次回溯累加即可，比较讨厌的是，要注意处理0的情况，递归函数d(i,j,k,s)，其中i为指向当前*个数的开头字符，j指向第二个数的开头字符，k为第三个数的开头字符
3代码
/**
 * @param {string} num
 * @return {boolean}
 */
var isAdditiveNumber = function(num) {
    if (num.length < 3) return false;
    let i=0;
    for (let j=i+1; j<num.length-1;j++) {
        for (let k=j+1; k<num.length; k++) {
            if (d(i,j,k,num)) return true;
        }
    }
    return false;
    function d(i,j,k,s) {
        if (k === s.length) return true;
        if (s[i] === ‘0’ && j > i + 1) return false;  //*个数字不是0，但以0开头
        if (s[j] === ‘0’ && k > j + 1) return false;  //第二个数字不是0，但以0开头
        let num1 = parseInt(s.slice(i,j));
        let num2 = parseInt(s.slice(j,k));
        for (let m=k+1; m<=s.length; m++) {
            if (s[k] === ‘0’ && m > k+1) return false;  //第三个数字不是0，但以0开头
            let num3 = parseInt(s.slice(k,m));
            if (num1 + num2 === num3) {
                return d(j,k,m,s);
            } else if (num1 + num2 < num3) {    //若前两个数已经小于第三个数，则提前结束
                return false;
            }
        }
        return false;
    }
};

307. 区域和检索 – 数组可修改（JS实现）

307. 区域和检索 – 数组可修改（JS实现）
1 题目
给定一个整数数组 nums，求出数组从索引 i 到 j (i ≤ j) 范围内元素的总和，包含 i, j 两点。
update(i, val) 函数可以通过将下标为 i 的数值更新为 val，从而对数列进行修改。
示例:
Given nums = [1, 3, 5]
sumRange(0, 2) -> 9
update(1, 2)
sumRange(0, 2) -> 8
说明:
数组仅可以在 update 函数下进行修改。
你可以假设 update 函数与 sumRange 函数的调用次数是均匀分布的。
链接：https://leetcode-cn.com/problems/range-sum-query-mutable
2 思路
这道题用动态规划的方法来做即可，在更新数组元素值时，也要更新dp数组
3代码
/**
 * @param {number[]} nums
 */
var NumArray = function(nums) {
    this.d = [];
    this.nums = nums;
    let sum = 0;
    for (let i=0; i<nums.length; i++) {
        sum += nums[i];
        this.d[i] = sum;
    }
};
/**
 * @param {number} i
 * @param {number} val
 * @return {void}
 */
NumArray.prototype.update = function(i, val) {
    let diff = val – this.nums[i];
    for (let j=i; j<this.d.length; j++) {
        this.d[j] += diff;
    }
    this.nums[i] = val;
};
/**
 * @param {number} i
 * @param {number} j
 * @return {number}
 */
NumArray.prototype.sumRange = function(i, j) {
    return this.d[j] – (i > 0 ? this.d[i-1] : 0);
};
/**
 * Your NumArray object will be instantiated and called as such:
 * var obj = new NumArray(nums)
 * obj.update(i,val)
 * var param_2 = obj.sumRange(i,j)
 */

310. *小高度树（JS实现）

310. *小高度树（JS实现）
1 题目
对于一个具有树特征的无向图，我们可选择任何一个节点作为根。图因此可以成为树，在所有可能的树中，具有*小高度的树被称为*小高度树。给出这样的一个图，写出一个函数找到所有的*小高度树并返回他们的根节点。
格式
该图包含 n 个节点，标记为 0 到 n – 1。给定数字 n 和一个无向边 edges 列表（每一个边都是一对标签）。
你可以假设没有重复的边会出现在 edges 中。由于所有的边都是无向边， [0, 1]和 [1, 0] 是相同的，因此不会同时出现在 edges 里。
示例 1:
输入: n = 4, edges = [[1, 0], [1, 2], [1, 3]]
0
|
1
/
2 3
输出: [1]
示例 2:
输入: n = 6, edges = [[0, 3], [1, 3], [2, 3], [4, 3], [5, 4]]
0 1 2
\ | /
3
|
4
|
5
输出: [3, 4]
说明:
根据树的定义，树是一个无向图，其中任何两个顶点只通过一条路径连接。 换句话说，一个任何没有简单环路的连通图都是一棵树。
树的高度是指根节点和叶子节点之间*长向下路径上边的数量。
链接：https://leetcode-cn.com/problems/minimum-height-trees
2 思路
这道题考察图的遍历，每次从图的*外层节点开始，删除图外层节点（入度为1的节点），直到图中没有节点，*后一次删除的节点就是*内层的根节点
3代码
/**
 * @param {number} n
 * @param {number[][]} edges
 * @return {number[]}
 */
var findMinHeightTrees = function(n, edges) {
    if (edges.length === 0) {   //对于没有边的情况，每个节点都是根节点
        let res = [];
        for (let i=0; i<n; i++) {
            res.push(i);
        }
        return res;
    }
    const map = {};
    for (let edge of edges) {    //建立图结构
        if (!map[edge[0]]) {
            map[edge[0]] = [];
            map[edge[0]].len = 0;
        }
        if (!map[edge[1]]) {
            map[edge[1]] = [];
            map[edge[1]].len = 0;
        }
        map[edge[0]].push(edge[1]);
        map[edge[1]].push(edge[0]);
        map[edge[0]].len++;
        map[edge[1]].len++;
    }
    let arr = Object.keys(map).filter(key => map[key].len === 1);
    let temp = [];
    let visited = {};
    let res = arr.slice();
    while(arr.length > 0 || temp.length > 0) {
        if (arr.length === 0) {
            arr = temp;
            temp = [];
            res = arr.slice();
        }
        let p = arr.shift();
        if (visited[p]) continue;
        visited[p] = true;
        for (let edge of map[p]) {   //将下一次外层节点推入
            if (–map[edge].len === 1) temp.push(edge);
        }
    }
    return res;
};

309. *佳买卖股票时机含冷冻期（JS实现）

309. *佳买卖股票时机含冷冻期（JS实现）
1 题目
给定一个整数数组，其中第 i 个元素代表了第 i 天的股票价格 。​
设计一个算法计算出*大利润。在满足以下约束条件下，你可以尽可能地完成更多的交易（多次买卖一支股票）:
你不能同时参与多笔交易（你必须在再次购买前出售掉之前的股票）。
卖出股票后，你无法在第二天买入股票 (即冷冻期为 1 天)。
示例:
输入: [1,2,3,0,2]
输出: 3
解释: 对应的交易状态为: [买入, 卖出, 冷冻期, 买入, 卖出]
链接：https://leetcode-cn.com/problems/best-time-to-buy-and-sell-stock-with-cooldown
2 思路
这道题是股票系列的变种, 动态规划d[i][j]，代表第i天持有状态为j的*大利润，状态转移方程d[i][0] = Math.max(d[i-1][0], d[i-1][1] + prices[i])、d[i][1] = Math.max(d[i-1][1], d[i-2][0] – prices[i])，注意由于股票有冷冻期，因此只能买入第i-2天的股票
3代码
/**
 * @param {number[]} prices
 * @return {number}
 */
var maxProfit = function(prices) {
    if (prices.length === 0) return 0;
    const d = [];
    const min = Number.MIN_SAFE_INTEGER;
    for (let i=0; i<prices.length; i++) {
        if (!d[i]) d[i] = [];
        d[i][0] = 0;
        d[i][1] = 0;
    }
    for (let i=0; i<prices.length; i++) {
        if (i === 0) {   //*天特殊处理
            d[0][0] = 0;
            d[0][1] = -prices[i];
            continue;
        }
        if (i === 1) {  //第二天情况特殊处理
            d[i][0] = Math.max(d[i-1][0], d[i-1][1] + prices[i]);
            d[i][1] = Math.max(d[i-1][1], -prices[i]);
            continue;
        }
        d[i][0] = Math.max(d[i-1][0], d[i-1][1] + prices[i]);
        d[i][1] = Math.max(d[i-1][1], d[i-2][0] – prices[i]);
    }
    return d[prices.length-1][0];
};

. *大单词长度乘积（JS实现）

18. *大单词长度乘积（JS实现）
1 题目
给定一个字符串数组 words，找到 length(word[i]) * length(word[j]) 的*大值，并且这两个单词不含有公共字母。你可以认为每个单词只包含小写字母。如果不存在这样的两个单词，返回 0。
示例 1:
输入: [“abcw”,“baz”,“foo”,“bar”,“xtfn”,“abcdef”]
输出: 16
解释: 这两个单词为 “abcw”, “xtfn”。
示例 2:
输入: [“a”,“ab”,“abc”,“d”,“cd”,“bcd”,“abcd”]
输出: 4
解释: 这两个单词为 “ab”, “cd”。
示例 3:
输入: [“a”,“aa”,“aaa”,“aaaa”]
输出: 0
解释: 不存在这样的两个单词。
链接：https://leetcode-cn.com/problems/maximum-product-of-word-lengths
2 思路
这道题的关键点在于如何快速判断两个单词是否有公共字符，因此就想到了利用位运算，我们可以将一个单词转换为数字，a代表1，b代表2，c代表4依次类推，注意相同的字符只加一次，判断两个单词是否有公共字符时只需将两个数字作&运算看结果是否为0即可，当然可能有多个单词转换后的数字相同，但我们只需要*长单词的长度即可
3代码
/**
 * @param {string[]} words
 * @return {number}
 */
var maxProduct = function(words) {
    let map = {};
    for (let word of words) {
        let bit = count(word);
        if (map[bit]) {  //只保留*大长度
            map[bit] = Math.max(map[bit], word.length);
        } else {
            map[bit] = word.length;
        }
    }
    let wordBits = Object.keys(map);
    let max = 0;
    for (let i=0; i<wordBits.length; i++) {
        for (let j=i+1; j<wordBits.length; j++) {
            if ((parseInt(wordBits[i]) & parseInt(wordBits[j])) === 0) {
                max = Math.max(max, map[wordBits[i]] * map[wordBits[j]]);
            }
        }
    }
    return max;
    function count(word) {   //计算单词的对应的数字
        let sum = 0;
        let startIndex = ‘a’.charCodeAt(0);
        let map1 = {};
        for (let alpha of word) {
            if (map1[alpha]) continue;
            map1[alpha] = true;
            sum += 1 << (alpha.charCodeAt() – startIndex);
        }
        return sum;
    }
};

用PHP命令行控制脚本

可执行文件

所有的PHP发行版，不论是编译自源代码的版本还是预创建的版本，都在默认情况下带有一个PHP可执行文件。这个可执行文件可以被用来运行命令行的PHP程序。

要在你的系统上找到这个可执行文件，就要遵照下面的步骤：

在Windows操作系统里，它被放在PHP主安装目录下，文件名是php.exe或者（在老版本的PHP里）是php-cli.exe。

在Linux操作系统里，它被保存在PHP安装目录的bin/子目录下。

不论是在哪一个操作系统里，你都需要对它进行测试，以保证它能够正常运行，方法是用-v参数调用它：

shell> /path/to/php -v
PHP 5.0.0 (cli) (built: Jun 1 2005 18:32:10)
Copyright (c) 1997-2004 The PHP Group
Zend Engine v2.0.0, Copyright (c) 1998-2004 Zend Technologies

它应该会返回PHP的版本号。

一个简单的PHP CLI程序
一旦找到了这个CLI可执行文件，你就可以用一个简单的程序来使用一下。创建一个简单的文本文件，其中包含有以下PHP代码，并把它保存为hello.php：

<?php
echo “Hello from the CLI”;
?>

现在，试着在命令行提示符下运行这个程序，方法是调用CLI可执行文件并提供脚本的文件名：

shell> /path/to/phphello.php
Hello from the CLI

使用标准的输入和输出
PHP CLI会定义三个常量，以便让在命令行提示符下与解释器进行交互操作更加容易。这些常量见表格A。

表格A

常量 说明
STDIN 标准的输入设备
STDOUT 标准的输出设备
STDERR 标准的错误设备
你可以在自己的PHP脚本里使用这三个常量，以接受用户的输入，或者显示处理和计算的结果。要更好地理解这一点，可以看看下面的脚本（列表A）：

列表A
<?php
// ask for input
fwrite(STDOUT, “Enter your name: “);

// get input
$name = trim(fgets(STDIN));

// write input back
fwrite(STDOUT, “Hello, $name!”);
?>
Look what happens when you run it:
shell> /path/to/phphello.php
Enter your name: Joe
Hello, Joe!

在这个脚本里，fwrite()函数首先会向标准的输出设备写一条消息，询问用户的姓名。然后它会把从标准输入设备获得的用户输入信息读取到一个PHP变量里，并它把合并成为一个字符串。然后就用fwrite()把这个字符串打印输出到标准的输出设备上。

使用命令行自变量
在命令行里输入程序参数来更改其运行方式是很常见的做法。你也可以对CLI程序这样做。PHP CLI带有两个特殊的变量，专门用来达到这个目的：一个是$argv变量，它通过命令行把传递给PHP脚本的参数保存为单独的数组元素；另一个是$ argc变量，它用来保存$argv数组里元素的个数。

用PHP脚本编写一段读取$argv并处理它所含参数的代码是很简单的。试试列表B里的示例脚本，看看它是如何工作的：

列表B
<?php
print_r($argv);
?>

Run this script by passing it some arbitrary values, and check the output:

shell> /path/to/phptest.php chocolate 276 “killer tie, dude!”
Array
( [0] => test.php
[1] => chocolate
[2] => 276
[3] => killer tie, dude!
)

正如你可以从输出的结果看到的，传递给test.php的值会自动地作为数组元素出现在$argv里。要注意的是，$argvis的*个自变量总是脚本自己的名称。

下面是一个更加复杂的例子（列表C）：

列表C
<?php
// check for all required arguments
// first argument is always name of script!
if ($argc != 4) {
die(“Usage: book.php <check-in-date> <num-nights> <room-type>/n”);
}

// remove first argument
array_shift($argv);

// get and use remaining arguments
$checkin = $argv[0];
$nights = $argv[1];
$type = $argv[2];
echo “You have requested a $type room for $nights nights, checking in on $checkin. Thank you for your order!/n”;
?>

下面是其用法的示例：

shell> /path/to/phpbook.php 21/05/2005 7 single
You have requested a single room for 7 nights, checking in on 21/05/2005. Thank you for your order!

在这里，脚本首先会检查$argc，以确保自变量的数量符合要求。它然后会从$argv里提取出每一个自变量，把它们打印输出到标准的输出设备上。

注意：你可以用Console_Getopt PEAR类向PHP增加更加复杂的命令行参数。

使用CLI参数
除了用命令行传递PHP脚本参数，你还可以传递PHP CLI参数以更改其工作方式。表格B就是一些重要参数的列表：

表格B

参数 说明
-a 交互式运行Run interactively
-c path 从path读取php的.ini文件
-n 不用读取php的.ini文件就直接运行
-m 列出经过编译的模块
-i 显示有关PHP构建的信息
-l 检查PHP脚本的句法
-s 以彩色方式显示源代码
-w 显示去掉注释之后的源代码
-h 显示帮助
交互模式
你还可以以交互方式使用PHP CLI，也就是输入命令，马上获得结果。要得到这种效果，只需要使用一个参数调用CLI可执行文件就行了，就像下面这样：

shell> /path/to/php -a

你会看到一个空行，你可以在里面输入PHP代码。看看：

shell> /path/to/php -a
Interactive mode enabled
<?php
echo mktime();
1121187283
echo 2+2;
4
exit();
shell>

或者，你可以不使用-a参数就调用CLI可执行文件，直接输入完整的脚本或者代码段。用<Ctrl>-D来结束代码段，并让CLI来执行它。见下面的例子：

shell> /path/to/php
<?php
echo date(“d-M-Y h:i:s”, time());
?>
<Ctrl-D>
12-Jul-2005 06:54:04

这就是PHP的命令行，现在你应该已经对PHP CLI有了足够的了解，并开始使用它了。

Android 扩展屏幕 实现主屏副屏同步或者异步显示

android的扩展屏幕是通过Presentation 类实现的。

Presentation继承自 Dialog。

主屏显示的代码
主屏播放一个video3的视频 。video3在项目的 assets文件里 可以随便拷贝一个放进去。
import android.app.Activity;
import android.content.Context;
import android.content.DialogInterface;
import android.media.AudioManager;
import android.media.MediaPlayer;
import android.media.MediaRouter;
import android.net.Uri;
import android.os.Bundle;
import android.util.Log;
import android.view.Display;
import android.view.SurfaceHolder;
import android.view.SurfaceView;
import android.view.View;
import android.view.WindowManager;
import android.widget.ImageButton;
import com.xinyintai.guoruiyibushow.R;
import java.io.IOException;

/**
* 用于测试 android 异步播放。
* <p>
* 主界面 lvds 信号播放的界面。
*/

public class MainActivity extends Activity {
private MediaRouter mMediaRouter;
private DemoPresentation mPresentation;
int flag = 2;// 1 绿色同步 2 蓝色异步

private SurfaceView surfaceView;
private MediaPlayer mediaPlayer1;
ImageButton bt;
String TAG = “MainActivity”;

@Override
protected void onCreate(Bundle savedInstanceState) {
super.onCreate(savedInstanceState);
// 得到media router service.
mMediaRouter = (MediaRouter) getSystemService(Context.MEDIA_ROUTER_SERVICE);
setContentView(R.layout.activity_main);
//监听media rotues的改变
mMediaRouter.addCallback(MediaRouter.ROUTE_TYPE_LIVE_VIDEO, mMediaRouterCallback);
initMainPlay();//主屏幕
updatePresentation();//扩展屏幕
}

//主屏幕播放 video3视频
private void initMainPlay() {
mediaPlayer1 = new MediaPlayer();
surfaceView = this.findViewById(R.id.surface1);
bt = findViewById(R.id.iv_state);
surfaceView.getHolder().setKeepScreenOn(true);
surfaceView.getHolder().addCallback(new SurfaceViewLis());
mediaPlayer1.setOnCompletionListener(new MediaPlayer.OnCompletionListener() {
@Override
public void onCompletion(MediaPlayer mp) {
play();
}
});

bt.setOnClickListener(new View.OnClickListener() {
@Override
public void onClick(View v) {
if (flag == 1) {//绿色异步
try {
mPresentation.show();//异步让这个dialog展示
} catch (WindowManager.InvalidDisplayException ex) {
// Log.w(TAG, “Couldn’t show presentation! Display was removed in “+ “the meantime.”, ex);
mPresentation = null;
}
bt.setBackground(getResources().getDrawable(R.drawable.circle_blue));
flag = 2;
} else if (flag == 2) {//蓝色同步
mPresentation.dismiss();//同步让presentation消失
bt.setBackground(getResources().getDrawable(R.drawable.circle_green));
flag = 1;
}
}
});
}

public void play() {
mediaPlayer1.reset();
mediaPlayer1.setAudioStreamType(AudioManager.STREAM_MUSIC);

try {
mediaPlayer1.setDataSource(this, Uri.parse(“android.resource://” + “com.xinyintai.guoruiyibushow” + “/” + R.raw.video3));
// 把视频输出到SurfaceView上
mediaPlayer1.setDisplay(surfaceView.getHolder());
mediaPlayer1.prepare();
mediaPlayer1.start();
} catch (IOException e) {
e.printStackTrace();
}

}

private class SurfaceViewLis implements SurfaceHolder.Callback {
@Override
public void surfaceChanged(SurfaceHolder holder, int format, int width,
int height) {
}

@Override
public void surfaceCreated(SurfaceHolder holder) {

try {
play();
mediaPlayer1.seekTo(0);
} catch (IllegalArgumentException e) {
// TODO Auto-generated catch block
e.printStackTrace();
} catch (SecurityException e) {
// TODO Auto-generated catch block
e.printStackTrace();
} catch (IllegalStateException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}

@Override
public void surfaceDestroyed(SurfaceHolder holder) {

}

}

@Override
protected void onResume() {
// TODO Auto-generated method stub
super.onResume();
}

private final MediaRouter.SimpleCallback mMediaRouterCallback =
new MediaRouter.SimpleCallback() {
@Override
public void onRouteSelected(MediaRouter router, int type, MediaRouter.RouteInfo info) {
Log.e(TAG, “onRouteSelected: type=” + type + “, info=” + info);
updatePresentation();
}

@Override
public void onRouteUnselected(MediaRouter router, int type, MediaRouter.RouteInfo info) {
Log.e(TAG, “onRouteeUnselected: type=” + type + “, info=” + info);
updatePresentation();
}

@Override
public void onRoutePresentationDisplayChanged(MediaRouter router, MediaRouter.RouteInfo info) {
Log.e(TAG, “onRoutePresentationDisplayChanged: info=” + info);
updatePresentation();
}
};

//准备异步的界面
private void updatePresentation() {
// Get the current route and its presentation display.
MediaRouter.RouteInfo route = mMediaRouter.getSelectedRoute(MediaRouter.ROUTE_TYPE_LIVE_VIDEO);
Display presentationDisplay = route != null ? route.getPresentationDisplay() : null;

// 如果屏幕显示改变了 让当前的扩展屏消失
if (mPresentation != null && mPresentation.getDisplay() != presentationDisplay) {
//Log.i(TAG, “Dismissing presentation because the current route no longer “+ “has a presentation display.”);
mPresentation.dismiss();
mPresentation = null;
}

//创建一个扩展屏
if (mPresentation == null && presentationDisplay != null) {
Log.e(TAG, “Showing presentation on display: ” + presentationDisplay);
mPresentation = new DemoPresentation(this, presentationDisplay);
mPresentation.setOnDismissListener(mOnDismissListener);
mPresentation.show();
}

updateContents();
}

//更新主界面中的内容
private void updateContents() {

}

/**
* 监听 副屏界面的消失 更新主界面。
*/
private final DialogInterface.OnDismissListener mOnDismissListener =
new DialogInterface.OnDismissListener() {
@Override
public void onDismiss(DialogInterface dialog) {
if (dialog == mPresentation) {
//Log.i(TAG, “Presentation was dismissed.”);mPresentation = null;
updateContents();
}
}
};

}
主界面的布局 一个surfaceview用来播放视频。imagebutton用来切换同步异步的按钮。
<RelativeLayout xmlns:android=”http://schemas.android.com/apk/res/android”
xmlns:tools=”http://schemas.android.com/tools”
android:layout_width=”match_parent”
android:layout_height=”match_parent”
tools:context=”.demo.MainActivity” >

<SurfaceView
android:id=”@+id/surface1″
android:layout_width=”match_parent”
android:layout_height=”wrap_content”
android:layout_centerInParent=”true” />

<!–切换同步异步的按钮的小按钮–>
<ImageButton
android:id=”@+id/iv_state”
android:visibility=”visible”
android:layout_alignParentLeft=”true”
android:layout_alignParentBottom=”true”
android:layout_marginLeft=”5dp”
android:layout_marginBottom=”5dp”
android:background=”@drawable/circle_green”
android:layout_width=”50dp”
android:layout_height=”50dp” />

</RelativeLayout>

*imagebutton的按钮background的背景。改一个绿色和蓝色的颜色即可

<?xml version=”1.0″ encoding=”utf-8″?>
<shape xmlns:android=”http://schemas.android.com/apk/res/android”
android:shape=”oval”
android:useLevel=”false”
>
<size android:width=”21dp”
android:height=”21dp” />
<stroke
android:width=”0dp”
android:color=”@android:color/white” />
<solid
android:color=”@color/colorPrimaryDark” />
</shape>

副屏的展示界面
副屏是用HDMI给的信号。
副屏的播放一个video的视频 。video在项目的 assets文件里 可以随便拷贝一个放进去。
import android.app.Presentation;
import android.content.Context;
import android.media.AudioManager;
import android.media.MediaPlayer;
import android.net.Uri;
import android.os.Bundle;
import android.util.Log;
import android.view.Display;
import android.view.SurfaceHolder;
import android.view.SurfaceView;
import com.xinyintai.guoruiyibushow.R;
import java.io.IOException;
import static android.content.ContentValues.TAG;

/**
* 用于测试 android 异步播放。
* <p>
* 副屏界面 hdmi信号口
*/

public class DemoPresentation extends Presentation {
private SurfaceView surfaceView;
private MediaPlayer mediaPlayer1;
boolean isLeave = true;

public DemoPresentation(Context context, Display display) {
super(context, display);
}

@Override
protected void onCreate(Bundle savedInstanceState) {
super.onCreate(savedInstanceState);
setContentView(R.layout.presentation_content);
// Set up the surface view for visual interest.
mediaPlayer1 = new MediaPlayer();
surfaceView = this.findViewById(R.id.surface2);
surfaceView.getHolder().setKeepScreenOn(true);
surfaceView.getHolder().addCallback(new SurfaceViewLis());

mediaPlayer1.setOnCompletionListener(new MediaPlayer.OnCompletionListener() {
@Override
public void onCompletion(MediaPlayer mp) {
if (isLeave) {
play(“android.resource://” + “com.xinyintai.guoruiyibushow” + “/” + R.raw.video);
}
}
});
}

public void pause() {
mediaPlayer1.pause();
}

public void play(String path) {
if (surfaceView != null) {
mediaPlayer1.reset();
mediaPlayer1.setAudioStreamType(AudioManager.STREAM_MUSIC);
try {
mediaPlayer1.setDataSource(getContext(), Uri.parse(path));
// 把视频输出到SurfaceView上
mediaPlayer1.setDisplay(surfaceView.getHolder());
mediaPlayer1.prepare();
mediaPlayer1.start();
} catch (IOException e) {
e.printStackTrace();
}
}

}

private class SurfaceViewLis implements SurfaceHolder.Callback {
@Override
public void surfaceChanged(SurfaceHolder holder, int format, int width,
int height) {
}

@Override
public void surfaceCreated(SurfaceHolder holder) {
Log.e(TAG, “surfaceCreated: surfaceivew产生了”);
try {
play(“android.resource://” + “com.xinyintai.guoruiyibushow” + “/” + R.raw.video);
mediaPlayer1.seekTo(0);
} catch (IllegalArgumentException e) {
// TODO Auto-generated catch block
e.printStackTrace();
} catch (SecurityException e) {
// TODO Auto-generated catch block
e.printStackTrace();
} catch (IllegalStateException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}

@Override
public void surfaceDestroyed(SurfaceHolder holder) {

}

}

@Override
public void onDetachedFromWindow() {
super.onDetachedFromWindow();
isLeave = false;
Log.e(TAG, “onDetachedFromWindow: 副屏消失了”);
}

@Override
public void onAttachedToWindow() {
super.onAttachedToWindow();
Log.e(TAG, “onAttachedToWindow: 副屏显示了”);
isLeave = true;
}

@Override
protected void onStop() {
super.onStop();
}
}

副屏的布局只有一个surfaceview用来播放视频
<RelativeLayout xmlns:android=”http://schemas.android.com/apk/res/android”
xmlns:tools=”http://schemas.android.com/tools”
android:layout_width=”match_parent”
android:layout_height=”match_parent”
tools:context=”.demo.MainActivity” >

<SurfaceView
android:id=”@+id/surface2″
android:layout_width=”match_parent”
android:layout_height=”wrap_content”
android:layout_centerInParent=”true” />

</RelativeLayout>

这样就实现了主屏和副屏同时播放视频。可以在主屏通过按钮。切换同步或者异步播放的功能。
————————————————

php 获取系统命令结果,php获取linux命令结果的实例代码

下面小编就为大家带来一篇php获取linux命令结果的实例。小编觉得挺不错的，现在就分享给大家，也给大家做个参考。一起跟随小编过来看看吧

如果使用php命令行里想获取etho网卡的IP怎么处理呢 ？

public function get_server_ip() {

if (PHP_SAPI === ‘cli’){

$buffer = “”;

$handle = popen(“ifconfig eth0|grep ‘inet addr’|awk -F'[ :]’ ‘{print $13}'”, ‘r’);

while(!feof($handle)) {

$buffer.=fgets($handle);

}

pclose($handle);

$server_ip = rtrim($buffer);

}else{

if (isset($_SERVER)) {

if($_SERVER[‘SERVER_ADDR’]) {

$server_ip = $_SERVER[‘SERVER_ADDR’];

} else {

$server_ip = $_SERVER[‘LOCAL_ADDR’];

}

} else {

$server_ip = getenv(‘SERVER_ADDR’);

}

}

return $server_ip;

}

php中判断shell_exec执行结果

在做第三方登录的时候，注册时头像下载比较耗时，所以改为了异步 shell_exec 执行wget

不过shell_exec执行成功无输出 和 执行失败 返回的都是null 在记录错误日志的时候区分就成问题了。

在php手册的评论中找到了区分的方法，在此记录一下。

$shell = “wget -O despath sourcepath && echo ‘success’ “;
$shellExec = shell_exec($shell);
var_dump($shellExec);

这样 当前面的执行成功的时候 会执行echo 执行结果就是success 而不是空了

执行失败不再继续执行echo 执行结果仍旧是null

这样就区分除了执行结果，记录对应日志就OK了。

补充一下，大概是这样的。。。

$shellExec = shell_exec(“cd ../”);
var_dump($shellExec);//NULL

$shellExec = shell_exec(“cd ../ && echo ‘suc'”);
var_dump($shellExec);//string(4) “suc ”

 

10分钟带你逆袭kafka之路

1. kafka概述

##1.1 kafka简介

Apache Kafka 是一个快速、可扩展的、高吞吐的、可容错的分布式“发布-订阅”消息系统， 使用 Scala 与 Java 语言编写，能够将消息从一个端点传递到另一个端点，较之传统的消息中 间件(例如 ActiveMQ、RabbitMQ)，Kafka 具有高吞吐量、内置分区、支持消息副本和高容 错的特性，非常适合大规模消息处理应用程序。

Kafka 官网: http://kafka.apache.org/

Kafka主要设计目标如下：

  • 以时间复杂度为O(1)的方式提供消息持久化能力，即使对TB级以上数据也能保证常数时间的访问性能。
  • 高吞吐率。即使在非常廉价的商用机器上也能做到单机支持每秒100K条消息的传输。
  • 支持Kafka Server间的消息分区，及分布式消费，同时保证每个partition内的消息顺序传输。
  • 同时支持离线数据处理和实时数据处理。
  • 支持在线水平扩展。

Kafka通常用于两大类应用程序：

  • 建立实时流数据管道，以可靠地在系统或应用程序之间获取数据
  • 构建实时流应用程序，以转换或响应数据流

要了解Kafka如何执行这些操作，让我们从头开始深入研究Kafka的功能。

首先几个概念：

  • Kafka在一个或多个可以跨越多个数据中心的服务器上作为集群运行。
  • Kafka集群将记录流存储在称为主题的类别中。
  • 每个记录由一个键，一个值和一个时间戳组成

1.2 kafka架构体系

在这里插入图片描述

1.3 kafka的应用场景

kafka的应用场景非常多, 下面我们就来举几个我们*常见的场景

1.3.1 用户的活动跟踪

用户在网站的不同活动消息发布到不同的主题中心，然后可以对这些消息进行实时监测、实时处理。当然，也可以加载到Hadoop或离线处理数据仓库，对用户进行画像。像淘宝、天猫、京东这些大型电商平台，用户的所有活动都要进行追踪的。

1.3.2 日志收集

在这里插入图片描述

1.3.3 限流削峰

在这里插入图片描述

1.3.4 高吞吐率实现

Kafka与其他MQ相比，*大的特点就是高吞吐率。为了增加存储能力，Kafka将所有的消息都写入到了低速大容量的硬盘。按理说，这将导致性能损失，但实际上，Kafka仍然可以保持超高的吞吐率，并且其性能并未受到影响。其主要采用如下方式实现了高吞吐率。

  1. 顺序读写：Kafka将消息写入到了分区partition中，而分区中的消息又是顺序读写的。顺序读写要快于随机读写。
  2. 零拷贝：生产者、消费者对于Kafka中的消息是采用零拷贝实现的。
  3. 批量发送：Kafka允许批量发送模式。
  4. 消息压缩：Kafka允许对消息集合进行压缩。

1.4 kafka的优点

1. 解耦：

在项目启动之初来预测将来项目会碰到什么需求，是*其困难的。消息系统在处理过程中间插入了一个隐含的、基于数据的接口层，两边的处理过程都要实现这一接口。这允许你独立的扩展或修改两边的处理过程，只要确保它们遵守同样的接口约束。

2. 冗余:(副本)

有些情况下，处理数据的过程会失败。除非数据被持久化，否则将造成丢失。消息队列把数据进行持久化直到它们已经被完全处理，通过这一方式规避了数据丢失风险。许多消息队列所采用的”插入-获取-删除”范式中，在把一个消息从队列中删除之前，需要你的处理系统明确的指出该消息已经被处理完毕，从而确保你的数据被安全的保存直到你使用完毕。

3. 扩展性

因为消息队列解耦了你的处理过程，所以增大消息入队和处理的频率是很容易的，只要另外增加处理过程即可。不需要改变代码、不需要调节参数。扩展就像调大电力按钮一样简单。

4. 灵活性&峰值处理能力

在访问量剧增的情况下，应用仍然需要继续发挥作用，但是这样的突发流量并不常见；如果为以能处理这类峰值访问为标准来投入资源随时待命无疑是巨大的浪费。使用消息队列能够使关键组件顶住突发的访问压力，而不会因为突发的超负荷的请求而完全崩溃。

5. 可恢复性

系统的一部分组件失效时，不会影响到整个系统。消息队列降低了进程间的耦合度，所以即使一个处理消息的进程挂掉，加入队列中的消息仍然可以在系统恢复后被处理。

6. 顺序保证

在大多使用场景下，数据处理的顺序都很重要。大部分消息队列本来就是排序的，并且能保证数据会按照特定的顺序来处理。Kafka保证一个Partition内的消息的有序性。

7. 缓冲

在任何重要的系统中，都会有需要不同的处理时间的元素。例如，加载一张图片比应用过滤器花费更少的时间。消息队列通过一个缓冲层来帮助任务*高效率的执行———写入队列的处理会尽可能的快速。该缓冲有助于控制和优化数据流经过系统的速度。

8. 异步通信

很多时候，用户不想也不需要立即处理消息。消息队列提供了异步处理机制，允许用户把一个消息放入队列，但并不立即处理它。想向队列中放入多少消息就放多少，然后在需要的时候再去处理它们。

1.5 kafka于其他MQ对比

1. RabbitMQ

RabbitMQ是使用Erlang编写的一个开源的消息队列，本身支持很多的协议：AMQP，XMPP, SMTP, STOMP，也正因如此，它非常重量级，更适合于企业级的开发。同时实现了Broker构架，这意味着消息在发送给客户端时先在中心队列排队。对路由，负载均衡或者数据持久化都有很好的支持。

2. Redis

Redis是一个基于Key-Value对的NoSQL数据库，开发维护很活跃。虽然它是一个Key-Value数据库存储系统，但它本身支持MQ功能，所以完全可以当做一个轻量级的队列服务来使用。对于RabbitMQ和Redis的入队和出队操作，各执行100万次，每10万次记录一次执行时间。测试数据分为128Bytes、512Bytes、1K和10K四个不同大小的数据。实验表明：入队时，当数据比较小时Redis的性能要高于RabbitMQ，而如果数据大小超过了10K，Redis则慢的无法忍受；出队时，无论数据大小，Redis都表现出非常好的性能，而RabbitMQ的出队性能则远低于Redis。

3. ZeroMQ

ZeroMQ号称*快的消息队列系统，尤其针对大吞吐量的需求场景。ZeroMQ能够实现RabbitMQ不擅长的高级/复杂的队列，但是开发人员需要自己组合多种技术框架，技术上的复杂度是对这MQ能够应用成功的挑战。ZeroMQ具有一个独特的非中间件的模式，你不需要安装和运行一个消息服务器或中间件，因为你的应用程序将扮演这个服务器角色。你只需要简单的引用ZeroMQ程序库，可以使用NuGet安装，然后你就可以愉快的在应用程序之间发送消息了。但是ZeroMQ仅提供非持久性的队列，也就是说如果宕机，数据将会丢失。其中，Twitter的Storm 0.9.0以前的版本中默认使用ZeroMQ作为数据流的传输（Storm从0.9版本开始同时支持ZeroMQ和Netty作为传输模块）。

4. ActiveMQ

ActiveMQ是Apache下的一个子项目。 类似于ZeroMQ，它能够以代理人和点对点的技术实现队列。同时类似于RabbitMQ，它少量代码就可以高效地实现高级应用场景。

5. Kafka/Jafka

Kafka是Apache下的一个子项目，是一个高性能跨语言分布式发布/订阅消息队列系统，而Jafka是在Kafka之上孵化而来的，即Kafka的一个升级版。具有以下特性：快速持久化，可以在O(1)的系统开销下进行消息持久化；高吞吐，在一台普通的服务器上既可以达到10W/s的吞吐速率；完全的分布式系统，Broker、Producer、Consumer都原生自动支持分布式，自动实现负载均衡；支持Hadoop数据并行加载，对于像Hadoop的一样的日志数据和离线分析系统，但又要求实时处理的限制，这是一个可行的解决方案。Kafka通过Hadoop的并行加载机制统一了在线和离线的消息处理。Apache Kafka相对于ActiveMQ是一个非常轻量级的消息系统，除了性能非常好之外，还是一个工作良好的分布式系统。

1.6 kafka的几种重要角色

1.6.1 kafka作为存储系统

任何允许发布与使用无关的消息发布的消息队列都有效地充当了运行中消息的存储系统。Kafka的不同之处在于它是一个非常好的存储系统。

写入Kafka的数据将写入磁盘并进行复制以实现容错功能。Kafka允许生产者等待确认，以便直到完全复制并确保即使写入服务器失败的情况下写入也不会完成。

Kafka的磁盘结构可以很好地扩展使用-无论服务器上有50 KB还是50 TB的持久数据，Kafka都将执行相同的操作。

由于认真对待存储并允许客户端控制其读取位置，因此您可以将Kafka视为一种专用于高性能，低延迟提交日志存储，复制和传播的专用分布式文件系统。

1.6.2 kafka作为消息传递系统

Kafka的流概念与传统的企业消息传递系统相比如何？

传统上，消息传递具有两种模型：排队和发布-订阅。在队列中，一组使用者可以从服务器中读取内容，并且每条记录都将转到其中一个。在发布-订阅记录中广播给所有消费者。这两个模型中的每一个都有优点和缺点。排队的优势在于，它允许您将数据处理划分到多个使用者实例上，从而扩展处理量。不幸的是，队列不是多用户的—一次进程读取了丢失的数据。发布-订阅允许您将数据广播到多个进程，但是由于每条消息都传递给每个订阅者，因此无法扩展处理。

Kfka的消费者群体概念概括了这两个概念。与队列一样，使用者组允许您将处理划分为一组进程（使用者组的成员）。与发布订阅一样，Kafka允许您将消息广播到多个消费者组。

Kafka模型的优点在于，每个主题都具有这些属性-可以扩展处理范围，并且是多订阅者-无需选择其中一个。

与传统的消息传递系统相比，Kafka还具有更强的订购保证。

传统队列将记录按顺序保留在服务器上，如果多个使用者从队列中消费，则服务器将按记录的存储顺序分发记录。但是，尽管服务器按顺序分发记录，但是这些记录是异步传递给使用者的，因此它们可能在不同的使用者上乱序到达。这实际上意味着在并行使用的情况下会丢失记录的顺序。消息传递系统通常通过“专有使用者”的概念来解决此问题，该概念仅允许一个进程从队列中使用，但是，这当然意味着在处理中没有并行性。

Kafka做得更好。通过在主题内具有并行性（即分区）的概念，Kafka能够在用户进程池中提供排序保证和负载均衡。这是通过将主题中的分区分配给消费者组中的消费者来实现的，以便每个分区都由组中的一个消费者完全消费。通过这样做，我们确保使用者是该分区的唯一读取器，并按顺序使用数据。由于存在许多分区，因此仍然可以平衡许多使用者实例上的负载。但是请注意，使用者组中的使用者实例不能超过分区。

1.6.3 kafka用作流处理

仅读取，写入和存储数据流是不够的，目的是实现对流的实时处理。

在Kafka中，流处理器是指从输入主题中获取连续数据流，对该输入进行一些处理并生成连续数据流以输出主题的任何东西。

例如，零售应用程序可以接受销售和装运的输入流，并输出根据此数据计算出的重新订购和价格调整流。

可以直接使用生产者和消费者API进行简单处理。但是，对于更复杂的转换，Kafka提供了完全集成的Streams API。这允许构建执行非重要处理的应用程序，这些应用程序计算流的聚合或将流连接在一起。

该功能有助于解决此类应用程序所面临的难题：处理无序数据，在代码更改时重新处理输入，执行状态计算等。

流API建立在Kafka提供的核心原语之上：它使用生产者和使用者API进行输入，使用Kafka进行状态存储，并使用相同的组机制来实现流处理器实例之间的容错。

2. kafka中的关键术语解释

2.1 Topic

主题。在 Kafka 中，使用一个类别属性来划分消息的所属类，划分消息的这个类称为 topic。 topic 相当于消息的分类标签，是一个逻辑概念

物理上不同Topic的消息分开存储，逻辑上一个Topic的消息虽然保存于一个或多个broker上但用户只需指定消息的Topic即可生产或消费数据而不必关心数据存于何处

2.2 Partition

分区。topic 中的消息被分割为一个或多个 partition，其是一个物理概念，对应到系统上 就是一个或若干个目录。partition 内部的消息是有序的，但 partition 间的消息是无序的。

2.3 Segment

段。将 partition 进一步细分为了若干的 segment，每个 segment 文件的大小相等。

2.4 Broker

Kafka 集群包含一个或多个服务器，每个服务器节点称为一个 broker。

broker存储topic的数据。如果某topic有N个partition，集群有N个broker，那么每个broker存储该topic的一个partition。

如果某topic有N个partition，集群有(N+M)个broker，那么其中有N个broker存储该topic的一个partition，剩下的M个broker不存储该topic的partition数据。

如果某topic有N个partition，集群中broker数目少于N个，那么一个broker存储该topic的一个或多个partition。在实际生产环境中，尽量避免这种情况的发生，这种情况容易导致Kafka集群数据不均衡。

2.5 Producer

生产者, 即消息的发布者. 生产者将数据发布到他们选择的主题。生产者负责选择将哪个记录分配给主题中的哪个分区。即: 生产者生产的一条消息，会被写入到某一个 partition。

2.6 Consumer

消费者。可以从 broker 中读取消息。

一个消费者可以消费多个 topic 的消息

一个消费者可以消费同一个 topic 中的多个 partition 中的消息

一个 partiton 允许多个 consumer 同时消费

2.7 Consumer Group

consumer group 是 kafka 提供的可扩展且具有容错性的消费者机制。组内可以有多个消 费者，它们共享一个公共的 ID，即 group ID。组内的所有消费者协调在一起来消费订阅主题 的所有分区。

Kafka 保证同一个 consumer group 中只有一个 consumer 会消费某条消息，实际上，Kafka 保证的是稳定状态下每一个 consumer 实例只会消费某一个或多个特定的 partition，而某个 partition 的数据只会被某一个特定的 consumer 实例所消费。

下面我们用官网的一张图, 来标识consumer数量和partition数量的对应关系

由两台服务器组成的Kafka群集，其中包含四个带有两个使用者组的分区（P0-P3）。消费者组A有两个消费者实例，组B有四个。

在这里插入图片描述

其实对于这个消费组, 以前一直搞不明白, 我自己的总结是:

topic中的partitoin到group是发布订阅的通信方式，即一条topic的partition的消息会被所有的group消费，属于一对多模式；group到consumer是点对点通信方式，属于一对一模式。

举个例子: 不使用group的话，启动10个consumer消费一个topic，这10个consumer都能得到topic的所有数据，相当于这个topic中的任一条消息被消费10次。

使用group的话，连接时带上groupid，topic的消息会分发到10个consumer上，每条消息只被消费1次

2.8 Replizcas of partition

分区副本。副本是一个分区的备份，是为了防止消息丢失而创建的分区的备份。

2.9 Partition Leader

每个 partition 有多个副本，其中有且仅有一个作为 Leader，Leader 是当前负责消息读写 的 partition。即所有读写操作只能发生于 Leader 分区上。

2.10 Partition Follower

所有Follower都需要从Leader同步消息，Follower与Leader始终保持消息同步。Leader 与 Follower 的关系是主备关系，而非主从关系。

2.11 ISR

  • ISR，In-Sync Replicas，是指副本同步列表。 ISR列表是由Leader负责维护。
  • AR，Assigned Replicas，指某个 partition 的所有副本, 即已分配的副本列表。
  • OSR，Outof-Sync Replicas, 即非同步的副本列表。
  • AR = ISR + OSR

2. 12 offset

偏移量。每条消息都有一个当前Partition下唯一的64字节的offset，它是相当于当前分区*条消息的偏移量。

2.13 Broker Controller

Kafka集群的多个broker中，有一个会被选举controller，负责管理整个集群中partition和replicas的状态。

只有 Broker Controller 会向 zookeeper 中注册 Watcher，其他 broker 及分区无需注册。即 zookeeper 仅需监听 Broker Controller 的状态变化即可。

2.14 HW与LEO

  • HW，HighWatermark，高水位，表示 Consumer 可以消费到的*高 partition 偏移量。HW 保证了 Kafka 集群中消息的一致性。确切地说，是保证了 partition 的 Follower 与 Leader 间数 据的一致性。
  • LEO，Log End Offset，日志*后消息的偏移量。消息是被写入到 Kafka 的日志文件中的， 这是当前*后一个写入的消息在 Partition 中的偏移量。
  • 对于 leader 新写入的消息，consumer 是不能立刻消费的。leader 会等待该消息被所有 ISR 中的 partition follower 同步后才会更新 HW，此时消息才能被 consumer 消费。

我相信你看完上面的概念还是懵逼的, 好吧, 下面我们就用图来形象话的表示两者的关系吧。

在这里插入图片描述

2.15 zookeeper

Zookeeper 负责维护和协调 broker，负责 Broker Controller 的选举。

在 kafka0.9 之前版本，offset 是由 zk 负责管理的。

总结:zk 负责 Controller 的选举，Controller 负责 leader 的选举。

2.16 Coordinator

Coordinator一般指的是运行在每个broker上的group Coordinator进程，用于管理Consumer Group中的各个成员，主要用于offset位移管理和Rebalance。一个Coordinator可以同时管理多个消费者组。

2. 17 Rebalance

当消费者组中的数量发生变化，或者topic中的partition数量发生了变化时，partition的所有权会在消费者间转移，即partition会重新分配，这个过程称为再均衡Rebalance。

再均衡能够给消费者组及broker带来高性能、高可用性和伸缩，但在再均衡期间消费者是无法读取消息的，即整个broker集群有小一段时间是不可用的。因此要避免不必要的再均衡。

2.18 offset commit

Consumer从broker中取一批消息写入buffer进行消费，在规定的时间内消费完消息后，会自动将其消费消息的offset提交给broker，以记录下哪些消息是消费过的。当然，若在时限内没有消费完毕，其是不会提交offset的。

3. kafka的工作原理和过程

3.1 消息写入算法

消息发送者将消息发送给broker, 并形成*终的可供消费者消费的log, 是已给比较复杂的过程：

  • producer先从zookeeper中找到该partition的leader
  • producer将消息发送给该leader
  • leader将消息接入本地的log, 并通知ISR的followers
  • ISR中的followers从leader中pull消息, 写入本地log后向leader发送ack
  • leader收到所有ISR中的followers的ack后, 增加HW并向producer发送ack, 表示消息写入成功

3.2 消息路由策略

在通过 API 方式发布消息时，生产者是以 Record 为消息进行发布的。Record 中包含 key 与 value，value 才是我们真正的消息本身，而 key 用于路由消息所要存放的 Partition。消息 要写入到哪个 Partition 并不是随机的，而是有路由策略的。

  • 若指定了 partition，则直接写入到指定的 partition;
  • 若未指定 partition 但指定了 key，则通过对 key 的 hash 值与 partition 数量取模，该取模
  • 结果就是要选出的 partition 索引;
  • 若 partition 和 key 都未指定，则使用轮询算法选出一个 partition。

3.3 HW截断机制

如果 partition leader 接收到了新的消息， ISR 中其它 Follower 正在同步过程中，还未同 步完毕时 leader 宕机。此时就需要选举出新的 leader。若没有 HW 截断机制，将会导致 partition 中 leader 与 follower 数据的不一致。

当原 Leader 宕机后又恢复时，将其 LEO 回退到其宕机时的 HW，然后再与新的 Leader进行数据同步，这样就可以保证老 Leader 与新 Leader 中数据一致了，这种机制称为 HW 截断机制。

3.4 消息发送的可靠性

生产者向 kafka 发送消息时，可以选择需要的可靠性级别。通过 request.required.acks参数的值进行设置。

1、0值

异步发送。生产者向 kafka 发送消息而不需要 kafka 反馈成功 ack。该方式效率*高，但可靠性*低。其可能会存在消息丢失的情况。

  • 在传输过程中会出现消息丢失。
  • 在broker内部会出现消息丢失。
  • 会出现写入到kafka中的消息的顺序与生产顺序不一致的情况。

2、1值

同步发送。生产者发送消息给 kafka，broker 的 partition leader 在收到消息后马上发送 成功 ack(无需等等 ISR 中的 Follower 同步)，生产者收到后知道消息发送成功，然后会再发送消息。如果一直未收到 kafka 的 ack，则生产者会认为消息发送失败，会重发消息。

该方式对于 Producer 来说，若没有收到 ACK，一定可以确认消息发送失败了，然后可以 重发;但是，即使收到了 ACK，也不能保证消息一定就发送成功了。故，这种情况，也可能 会发生消息丢失的情况。

3、-1值

同步发送。生产者发送消息给 kafka，kafka 收到消息后要等到 ISR 列表中的所有副本都 同步消息完成后，才向生产者发送成功 ack。如果一直未收到 kafka 的 ack，则认为消息发送 失败，会自动重发消息。该方式会出现消息重复接收的情况。

3.5 消费者消费过程解析

生产者将消息发送到topitc中, 消费者即可对其进行消费, 其消费过程如下:

  1. consumer向broker提交连接请求，其所连接上的broker都会向其发送broker controller的通信URL，即配置文件中的listeners地址；
  2. 当consumer指定了要消费的topic后，会向broker controller发送消费请求；
  3. broker controller会为consumer分配一个或几个partition leader，并将该partition的当前offset发送给consumer；
  4. consumer会按照broker controller分配的partition对其中的消息进行消费；
  5. 当consumer消费完该条消息后，consumer会向broker发送一个消息已经被消费反馈，即该消息的offset；
  6. 在broker接收到consumer的offset后，会更新相应的__consumer_offset中；

以上过程会一直重复，知道消费者停止请求消费；

Consumer可以重置offset，从而可以灵活消费存储在broker上的消息。

3.6 Partition Leader选举范围

当leader宕机后，broker controller会从ISR中挑选一个follower成为新的leader。如果ISR中没有其他副本怎么办？可以通过unclean.leader.election.enable的值来设置leader选举范围。

1、false

必须等到ISR列表中所有的副本都活过来才进行新的选举。该策略可靠性有保证，但可用性低。

2、true

在ISR列表中没有副本的情况下，可以选择任意一个没有宕机的主机作为新的leader，该策略可用性高，但可靠性没有保证。

3.7 重复消费问题的解决方案

1、同一个consumer重复消费

当Consumer由于消费能力低而引发了消费超时，则可能会形成重复消费。

在某数据刚好消费完毕，但是正准备提交offset时候，消费时间超时，则broker认为这条消息未消费成功。这时就会产生重复消费问题。

**其解决方案：**延长offset提交时间。

2、不同的consumer重复消费

当Consumer消费了消息，但还没有提交offset时宕机，则这些已经被消费过的消息会被重复消费。

**其解决方案：**将自动提交改为手动提交。

3.8 从架构设计上解决kafka重复消费的问题

其实在开发的时候, 我们在设计程序的时候, 比如考虑到网络故障等一些异常的情况, 我们都会设置消息的重试次数,

可能还有其他可能出现消息重复, 那我们应该如何解决呢?

下面提供三个方案:

3.8.1 方案一: 保存并查询

给每个消息都设置一个独一无二的uuid, 所有的消息, 我们都要存一个uuid, 我们在消费消息的时候, 首先去持久化系统中查询一下, 看这个看是否以前消费过, 如没有消费过, 在进行消费, 如果已经消费过, 丢弃就好了, 下图, 表明了这种方案：

在这里插入图片描述

3.8.2 方案二: 利用幂等

幂等（Idempotence）在数学上是这样定义的，如果一个函数 f(x) 满足：f(f(x)) = f(x)，则函数 f(x) 满足幂等性。

这个概念被拓展到计算机领域，被用来描述一个操作、方法或者服务。一个幂等操作的特点是，其任意多次执行所产生的影响均与一次执行的影响相同。一个幂等的方法，使用同样的参数，对它进行多次调用和一次调用，对系统产生的影响是一样的。所以，对于幂等的方法，不用担心重复执行会对系统造成任何改变。

我们举个例子来说明一下。在不考虑并发的情况下，“将 X 老师的账户余额设置为 100 万元”，执行一次后对系统的影响是，X 老师的账户余额变成了 100 万元。只要提供的参数 100万元不变，那即使再执行多少次，X 老师的账户余额始终都是 100万元，不会变化，这个操作就是一个幂等的操作。

再举一个例子，“将 X 老师的余额加 100 万元”，这个操作它就不是幂等的，每执行一次，账户余额就会增加 100 万元，执行多次和执行一次对系统的影响（也就是账户的余额）是不一样的。

所以，通过这两个例子，我们可以想到如果系统消费消息的业务逻辑具备幂等性，那就不用担心消息重复的问题了，因为同一条消息，消费一次和消费多次对系统的影响是完全一样的。也就可以认为，消费多次等于消费一次。

那么，如何实现幂等操作呢？*好的方式就是，从业务逻辑设计上入手，将消费的业务逻辑设计成具备幂等性的操作。但是，不是所有的业务都能设计成天然幂等的，这里就需要一些方法和技巧来实现幂等。

下面我们介绍一种常用的方法：利用数据库的唯一约束实现幂等。

例如，我们刚刚提到的那个不具备幂等特性的转账的例子：将 X 老师的账户余额加 100 万元。在这个例子中，我们可以通过改造业务逻辑，让它具备幂等性。

首先，我们可以限定，对于每个转账单每个账户只可以执行一次变更操作，在分布式系统中，这个限制实现的方法非常多，*简单的是我们在数据库中建一张转账流水表，这个表有三个字段：转账单 ID、账户 ID 和变更金额，然后给转账单 ID 和账户 ID 这两个字段联合起来创建一个唯一约束，这样对于相同的转账单 ID 和账户 ID，表里至多只能存在一条记录。

这样，我们消费消息的逻辑可以变为：“在转账流水表中增加一条转账记录，然后再根据转账记录，异步操作更新用户余额即可。”在转账流水表增加一条转账记录这个操作中，由于我们在这个表中预先定义了“账户 ID 转账单 ID”的唯一约束，对于同一个转账单同一个账户只能插入一条记录，后续重复的插入操作都会失败，这样就实现了一个幂等的操作。

在这里插入图片描述

3.8.3 方案三: 设置前提条件

为更新的数据设置前置条件另外一种实现幂等的思路是，给数据变更设置一个前置条件，如果满足条件就更新数据，否则拒*更新数据，在更新数据的时候，同时变更前置条件中需要判断的数据。

这样，重复执行这个操作时，由于*次更新数据的时候已经变更了前置条件中需要判断的数据，不满足前置条件，则不会重复执行更新数据操作。

比如，刚刚我们说过，“将 X 老师的账户的余额增加 100 万元”这个操作并不满足幂等性，我们可以把这个操作加上一个前置条件，变为：“如果X老师的账户当前的余额为 500万元，将余额加 100万元”，这个操作就具备了幂等性。

对应到消息队列中的使用时，可以在发消息时在消息体中带上当前的余额，在消费的时候进行判断数据库中，当前余额是否与消息中的余额相等，只有相等才执行变更操作。

但是，如果我们要更新的数据不是数值，或者我们要做一个比较复杂的更新操作怎么办？用什么作为前置判断条件呢？更加通用的方法是，给你的数据增加一个版本号属性，每次更数据前，比较当前数据的版本号是否和消息中的版本号一致，如果不一致就拒*更新数据，更新数据的同时将版本号 +1，一样可以实现幂等。

在这里插入图片描述

4 . kafka集群搭建

我们在工作中, 为了保证环境的高可用, 防止单点, kafka都是以集群的方式出现的, 下面就带领大家一起搭建一套kafka集群环境

我们在官网下载kafka, 下载地址为: http://kafka.apache.org/downloads, 下载我们需要的版本, 推荐使用稳定的版本

4.1 搭建集群

1、下载并解压

  1. cd /usr/local/src
  2. wget http://mirrors.tuna.tsinghua.edu.cn/apache/kafka/2.4.0/kafka_2.11–2.4.0.tgz
  3. mkdir /data/servers
  4. tar xzvf kafka_2.11–2.4.0.tgz -C /data/servers/
  5. cd /data/servers/kafka_2.11–2.4.0

2、修改配置文件

kafka的配置文件$KAFKA_HOME/config/server.properties, 主要修改一下下面几项：

  1. # 确保每个机器上的id不一样
  2. broker.id=0
  3. # 配置服务端的监控地址
  4. listeners=PLAINTEXT://192.168.51.128:9092
  5. # kafka 日志目录
  6. log.dirs=/data/servers/kafka_2.11–2.4.0/logs
  7. # kafka设置的partitons的个数
  8. num.partitions=1
  9. # zookeeper的连接地址, 如果有自己的zookeeper集群, 请直接使用自己搭建的zookeeper集群
  10. zookeeper.connect=192.168.51.128:2181

因为我自己是本机做实验, 所有使用的是一个主机的不同端口，在线上,，就是不同的机器，大家参考即可。

我们这里使用kafka的zookeeper，只启动一个节点， 但是正真的生产过程中，是需要zookeeper集群，自己搭建就好，后期我们也会出zookeeper的教程，大家请关注就好了。

3、拷贝3份配置文件

  1. # 创建对应的日志目录
  2. mkdir -p /data/servers/kafka_2.11–2.4.0/logs/9092
  3. mkdir -p /data/servers/kafka_2.11–2.4.0/logs/9093
  4. mkdir -p /data/servers/kafka_2.11–2.4.0/logs/9094
  6. # 拷贝三份配置文件
  7. cp server.properties server_9092.properties
  8. cp server.properties server_9093.properties
  9. cp server.properties server_9094.properties

修改不同端口对应的文件：

  1. # 9092的id为0, 9093的id为1, 9094的id为2
  2. broker.id=0
  3. # 配置服务端的监控地址, 分别在不通的配置文件中写入不同的端口
  4. listeners=PLAINTEXT://192.168.51.128:9092
  5. # kafka 日志目录, 目录也是对应不同的端口
  6. log.dirs=/data/servers/kafka_2.11–2.4.0/logs/9092
  7. # kafka设置的partitons的个数
  8. num.partitions=1
  9. # zookeeper的连接地址, 如果有自己的zookeeper集群, 请直接使用自己搭建的zookeeper集群
  10. zookeeper.connect=192.168.51.128:2181

4、修改zookeeper的配置文件

  1. dataDir=/data/servers/zookeeper
  2. server.1=192.168.51.128:2888:3888

然后创建zookeeper的myid文件

echo "1"> /data/servers/zookeeper/myid

5、启动zookeeper

使用kafka内置的zookeeper

  1. cd /data/servers/kafka_2.11–2.4.0/bin
  2. zookeeper-server-start.sh -daemon ../config/zookeeper.properties
  3. netstat -anp |grep 2181

6、启动kafka

  1. ./kafka-server-start.sh -daemon ../config/server_9092.properties
  2. ./kafka-server-start.sh -daemon ../config/server_9093.properties
  3. ./kafka-server-start.sh -daemon ../config/server_9094.properties

4.2 kafka的操作

1. topic

我们先来看一下创建topic常用的参数吧

–create 创建topic

–delete 删除topic

–alter 修改topic的名字或者partition个数

–list 查看topic

–describe 查看topic的详细信息

–topic <String: topic> 指定topic的名字

–zookeeper <String: hosts> 指定zookeeper的连接地址,

​ 参数提示并不赞成这样使用

​ DEPRECATED, The connection string for
​ the zookeeper connection in the form
​ host:port. Multiple hosts can be
​ given to allow fail-over.

–bootstrap-server <String: server to connect to>: 指定kafka的连接地址, 推荐使用这个,

​ 参数的提示信息显示

​ REQUIRED: The Kafka server to connect

  1. to. In case of providing this, a
  2. direct Zookeeper connection won‘t be
  3. required.

–replication-factor <Integer: replication factor> : 对于每个partiton的备份个数

​ The replication factor for each
​ partition in the topic being
​ created. If not supplied, defaults
​ to the cluster default.

–partitions <Integer: # of partitions>: 指定该topic的分区的个数

示例:

  1. cd /data/servers/kafka_2.11–2.4.0/bin
  2. # 创建topic test1
  3. kafka-topics.sh –create –bootstrap-server=192.168.51.128:9092,10.231.128.96:9093,192.168.51.128:9094 –replication-factor 1 –partitions 1 –topic test1
  4. # 创建topic test2
  5. kafka-topics.sh –create –bootstrap-server=192.168.51.128:9092,10.231.128.96:9093,192.168.51.128:9094 –replication-factor 1 –partitions 1 –topic test2
  6. # 查看topic
  7. kafka-topics.sh –list –bootstrap-server=192.168.51.128:9092,10.231.128.96:9093,192.168.51.128:9094

2、自动创建topic

我们在工作中, 如果我们不想去管理topic, 可以通过kafka的配置文件来管理, 我们可以让kafka自动创建topic, 需要在我们的kafka配置文件中加入如下配置文件

auto.create.topics.enable=true

如果删除topic想达到物理删除的目的, 也是需要配置的

delete.topic.enable=true

3、发送消息

他们可以通过客户端的命令生产消息

先来看看kafka-console-producer.sh常用的几个参数吧

–topic <String: topic> 指定topic

–timeout <Integer: timeout_ms> 超时时间

–sync 异步发送消息

–broker-list <String: broker-list> 官网提示: REQUIRED: The broker list string in the form HOST1:PORT1,HOST2:PORT2. 这个参数是必须的

kafka-console-producer.sh --broker-list 192.168.51.128:9092,192.168.51.128:9093,192.168.51.128:9094 --topic test1

4、消费消息

我们也还是先来看看kafka-console-consumer.sh的参数吧

–topic <String: topic> 指定topic

–group <String: consumer group id> 指定消费者组

–from-beginning : 指定从开始进行消费, 如果不指定, 就从当前进行消费

–bootstrap-server : kafka的连接地址

kafka-console-consumer.sh --bootstrap-server 192.168.51.128:9092,192.168.51.128:9093,192.168.51.128:9094 --topic test1 ---beginning

在这里插入图片描述

4.3 kafka的日志

kafka的日志分两种:

*种日志: 是我们的kafka的启动日志, 就是我们排查问题, 查看报错信息的日志,

第二种日志:就是我们的数据日志, kafka是我们的数据是以日志的形式存在存盘中的, 我们第二种所说的日志就是我们的partiton与segment

那我们就来说说备份和分区吧

我们创建一个分区, 一个备份, 那么test就应该在三台机器上或者三个数据目录只有一个test-0, (分区的下标是从0开始的)

如果我们创建N个分区, 我们就会在三个服务器上发现, test_0-n

如果我们创建M个备份, 我们就会在发现, test_0 到test_n 每一个都是M个

5. kafaka API

5.1 使用kafaka原生的api

1.消费者自动提交:

定义自己的生产者

  1. import org.apache.kafka.clients.producer.Callback;
  2. import org.apache.kafka.clients.producer.KafkaProducer;
  3. import org.apache.kafka.clients.producer.ProducerRecord;
  4. import org.apache.kafka.clients.producer.RecordMetadata;
  6. import java.util.Properties;
  8. /**
  9. * @ClassName MyKafkaProducer
  10. * @Description TODO
  11. * @Author lingxiangxiang
  12. * @Date 3:37 PM
  13. * @Version 1.0
  14. **/
  15. public class MyKafkaProducer {
  16. private org.apache.kafka.clients.producer.KafkaProducer<Integer, String> producer;
  18. public MyKafkaProducer() {
  19. Properties properties = new Properties();
  20. properties.put(“bootstrap.servers”, “192.168.51.128:9092,192.168.51.128:9093,192.168.51.128:9094”);
  21. properties.put(“key.serializer”, “org.apache.kafka.common.serialization.IntegerSerializer”);
  22. properties.put(“value.serializer”, “org.apache.kafka.common.serialization.StringSerializer”);
  23. // 设置批量发送
  24. properties.put(“batch.size”, 16384);
  25. // 批量发送的等待时间50ms, 超过50ms, 不足批量大小也发送
  26. properties.put(“linger.ms”, 50);
  27. this.producer = new org.apache.kafka.clients.producer.KafkaProducer<Integer, String>(properties);
  28. }
  30. public boolean sendMsg() {
  31. boolean result = true;
  32. try {
  33. // 正常发送, test2是topic, 0代表的是分区, 1代表的是key, hello world是发送的消息内容
  34. final ProducerRecord<Integer, String> record = new ProducerRecord<Integer, String>(“test2”, 0, 1, “hello world”);
  35. producer.send(record);
  36. // 有回调函数的调用
  37. producer.send(record, new Callback() {
  38. @Override
  39. public void onCompletion(RecordMetadata recordMetadata, Exception e) {
  40. System.out.println(recordMetadata.topic());
  41. System.out.println(recordMetadata.partition());
  42. System.out.println(recordMetadata.offset());
  43. }
  44. });
  45. // 自己定义一个类
  46. producer.send(record, new MyCallback(record));
  47. } catch (Exception e) {
  48. result = false;
  49. }
  50. return result;
  51. }
  52. }

生产者测试类:

在生产者测试类中,自己遇到一个坑, 就是*后自己没有加sleep, 就是怎么检查自己的代码都没有问题, 但是*后就是没法发送成功消息, *后加了一个sleep就可以了, 因为主函数main已经执行完退出, 但是消息并没有发送完成, 需要进行等待一下.当然, 你在生产环境中可能不会遇到这样问题, 呵呵, 代码如下：

  1. import static java.lang.Thread.sleep;
  3. /**
  4. * @ClassName MyKafkaProducerTest
  5. * @Description TODO
  6. * @Author lingxiangxiang
  7. * @Date 3:46 PM
  8. * @Version 1.0
  9. **/
  10. public class MyKafkaProducerTest {
  11. public static void main(String[] args) throws InterruptedException {
  12. MyKafkaProducer producer = new MyKafkaProducer();
  13. boolean result = producer.sendMsg();
  14. System.out.println(“send msg “ + result);
  15. sleep(1000);
  16. }
  17. }

消费者类：

  1. import kafka.utils.ShutdownableThread;
  2. import org.apache.kafka.clients.consumer.ConsumerRecord;
  3. import org.apache.kafka.clients.consumer.ConsumerRecords;
  4. import org.apache.kafka.clients.consumer.KafkaConsumer;
  6. import java.util.Arrays;
  7. import java.util.Collections;
  8. import java.util.Properties;
  10. /**
  11. * @ClassName MyKafkaConsumer
  12. * @Description TODO
  13. * @Author lingxiangxiang
  14. * @Date 4:12 PM
  15. * @Version 1.0
  16. **/
  17. public class MyKafkaConsumer extends ShutdownableThread {
  19. private KafkaConsumer<Integer, String> consumer;
  21. public MyKafkaConsumer() {
  22. super(“KafkaConsumerTest”, false);
  23. Properties properties = new Properties();
  24. properties.put(“bootstrap.servers”, “192.168.51.128:9092,192.168.51.128:9093,192.168.51.128:9094”);
  25. properties.put(“group.id”, “mygroup”);
  26. properties.put(“enable.auto.commit”, “true”);
  27. properties.put(“auto.commit.interval.ms”, “1000”);
  28. properties.put(“session.timeout.ms”, “30000”);
  29. properties.put(“heartbeat.interval.ms”, “10000”);
  30. properties.put(“auto.offset.reset”, “earliest”);
  31. properties.put(“key.deserializer”, “org.apache.kafka.common.serialization.IntegerDeserializer”);
  32. properties.put(“value.deserializer”, “org.apache.kafka.common.serialization.StringDeserializer”);
  33. this.consumer = new KafkaConsumer<Integer, String>(properties);
  34. }
  36. @Override
  37. public void doWork() {
  38. consumer.subscribe(Arrays.asList(“test2”));
  39. ConsumerRecords<Integer, String>records = consumer.poll(1000);
  40. for (ConsumerRecord record : records) {
  41. System.out.println(“topic = “ + record.topic());
  42. System.out.println(“partition = “ + record.partition());
  43. System.out.println(“key = “ + record.key());
  44. System.out.println(“value = “ + record.value());
  45. }
  46. }
  47. }

消费者的测试类：

  1. /**
  2. * @ClassName MyConsumerTest
  3. * @Description TODO
  4. * @Author lingxiangxiang
  5. * @Date 4:23 PM
  6. * @Version 1.0
  7. **/
  8. public class MyConsumerTest {
  9. public static void main(String[] args) {
  10. MyKafkaConsumer consumer = new MyKafkaConsumer();
  11. consumer.start();
  12. System.out.println(“==================”);
  13. }
  14. }

在这里插入图片描述

在这里插入图片描述

2. 消费者同步手动提交

前面的消费者都是以自动提交 offset 的方式对 broker 中的消息进行消费的，但自动提交 可能会出现消息重复消费的情况。所以在生产环境下，很多时候需要对 offset 进行手动提交， 以解决重复消费的问题。

手动提交又可以划分为同步提交、异步提交，同异步联合提交。这些提交方式仅仅是 doWork()方法不相同，其构造器是相同的。所以下面首先在前面消费者类的基础上进行构造 器的修改，然后再分别实现三种不同的提交方式。

同步提交方式是，消费者向 broker 提交 offset 后等待 broker 成功响应。若没有收到响 应，则会重新提交，直到获取到响应。而在这个等待过程中，消费者是阻塞的。其严重影响了消费者的吞吐量。

修改前面的MyKafkaConsumer.java, 主要修改下面的配置：

  1. import kafka.utils.ShutdownableThread;
  2. import org.apache.kafka.clients.consumer.ConsumerRecord;
  3. import org.apache.kafka.clients.consumer.ConsumerRecords;
  4. import org.apache.kafka.clients.consumer.KafkaConsumer;
  6. import java.util.Arrays;
  7. import java.util.Collections;
  8. import java.util.Properties;
  10. /**
  11. * @ClassName MyKafkaConsumer
  12. * @Description TODO
  13. * @Author lingxiangxiang
  14. * @Date 4:12 PM
  15. * @Version 1.0
  16. **/
  17. public class MyKafkaConsumer extends ShutdownableThread {
  19. private KafkaConsumer<Integer, String> consumer;
  21. public MyKafkaConsumer() {
  22. super(“KafkaConsumerTest”, false);
  23. Properties properties = new Properties();
  24. properties.put(“bootstrap.servers”, “192.168.51.128:9092,192.168.51.128:9093,192.168.51.128:9094”);
  25. properties.put(“group.id”, “mygroup”);
  26. // 这里要修改成手动提交
  27. properties.put(“enable.auto.commit”, “false”);
  28. // properties.put(“auto.commit.interval.ms”, “1000”);
  29. properties.put(“session.timeout.ms”, “30000”);
  30. properties.put(“heartbeat.interval.ms”, “10000”);
  31. properties.put(“auto.offset.reset”, “earliest”);
  32. properties.put(“key.deserializer”, “org.apache.kafka.common.serialization.IntegerDeserializer”);
  33. properties.put(“value.deserializer”, “org.apache.kafka.common.serialization.StringDeserializer”);
  34. this.consumer = new KafkaConsumer<Integer, String>(properties);
  35. }
  37. @Override
  38. public void doWork() {
  39. consumer.subscribe(Arrays.asList(“test2”));
  40. ConsumerRecords<Integer, String>records = consumer.poll(1000);
  41. for (ConsumerRecord record : records) {
  42. System.out.println(“topic = “ + record.topic());
  43. System.out.println(“partition = “ + record.partition());
  44. System.out.println(“key = “ + record.key());
  45. System.out.println(“value = “ + record.value());
  47. //手动同步提交
  48. consumer.commitSync();
  49. }
  51. }
  52. }

3、消费者异步手工提交

手动同步提交方式需要等待 broker 的成功响应，效率太低，影响消费者的吞吐量。异步提交方式是，消费者向 broker 提交 offset 后不用等待成功响应，所以其增加了消费者的吞吐量。

  1. import kafka.utils.ShutdownableThread;
  2. import org.apache.kafka.clients.consumer.ConsumerRecord;
  3. import org.apache.kafka.clients.consumer.ConsumerRecords;
  4. import org.apache.kafka.clients.consumer.KafkaConsumer;
  6. import java.util.Arrays;
  7. import java.util.Collections;
  8. import java.util.Properties;
  10. /**
  11. * @ClassName MyKafkaConsumer
  12. * @Description TODO
  13. * @Author lingxiangxiang
  14. * @Date 4:12 PM
  15. * @Version 1.0
  16. **/
  17. public class MyKafkaConsumer extends ShutdownableThread {
  19. private KafkaConsumer<Integer, String> consumer;
  21. public MyKafkaConsumer() {
  22. super(“KafkaConsumerTest”, false);
  23. Properties properties = new Properties();
  24. properties.put(“bootstrap.servers”, “192.168.51.128:9092,192.168.51.128:9093,192.168.51.128:9094”);
  25. properties.put(“group.id”, “mygroup”);
  26. // 这里要修改成手动提交
  27. properties.put(“enable.auto.commit”, “false”);
  28. // properties.put(“auto.commit.interval.ms”, “1000”);
  29. properties.put(“session.timeout.ms”, “30000”);
  30. properties.put(“heartbeat.interval.ms”, “10000”);
  31. properties.put(“auto.offset.reset”, “earliest”);
  32. properties.put(“key.deserializer”, “org.apache.kafka.common.serialization.IntegerDeserializer”);
  33. properties.put(“value.deserializer”, “org.apache.kafka.common.serialization.StringDeserializer”);
  34. this.consumer = new KafkaConsumer<Integer, String>(properties);
  35. }
  37. @Override
  38. public void doWork() {
  39. consumer.subscribe(Arrays.asList(“test2”));
  40. ConsumerRecords<Integer, String>records = consumer.poll(1000);
  41. for (ConsumerRecord record : records) {
  42. System.out.println(“topic = “ + record.topic());
  43. System.out.println(“partition = “ + record.partition());
  44. System.out.println(“key = “ + record.key());
  45. System.out.println(“value = “ + record.value());
  47. //手动同步提交
  48. // consumer.commitSync();
  49. //手动异步提交
  50. // consumer.commitAsync();
  51. // 带回调公共的手动异步提交
  52. consumer.commitAsync((offsets, e) -> {
  53. if(e != null) {
  54. System.out.println(“提交次数, offsets = “ + offsets);
  55. System.out.println(“exception = “ + e);
  56. }
  57. });
  58. }
  59. }
  60. }

5.2 springboot使用kafka

现在大家的开发过程中, 很多都用的是springboot的项目, 直接启动了, 如果还是用原生的API, 就是有点low了啊, 那kafka是如何和springboot进行联合的呢?

1. maven配置

  1. <!– https://mvnrepository.com/artifact/org.apache.kafka/kafka-clients –>
  2. <dependency>
  3. <groupId>org.apache.kafka</groupId>
  4. <artifactId>kafka-clients</artifactId>
  5. <version>2.1.1</version>
  6. </dependency>

2. 添加配置文件

在application.properties中加入如下配置信息:

  1. # kafka 连接地址
  2. spring.kafka.bootstrap-servers = 192.168.51.128:9092,10.231.128.96:9093,192.168.51.128:9094
  4. # 生产者
  5. spring.kafka.producer.acks = 0
  6. spring.kafka.producer.key-serializer = org.apache.kafka.common.serialization.StringSerializer
  7. spring.kafka.producer.value-serializer = org.apache.kafka.common.serialization.StringSerializer
  8. spring.kafka.producer.retries = 3
  9. spring.kafka.producer.batch-size = 4096
  10. spring.kafka.producer.buffer-memory = 33554432
  11. spring.kafka.producer.compression-type = gzip
  13. # 消费者
  14. spring.kafka.consumer.group-id = mygroup
  15. spring.kafka.consumer.auto-commit-interval = 5000
  16. spring.kafka.consumer.heartbeat-interval = 3000
  17. spring.kafka.consumer.key-deserializer = org.apache.kafka.common.serialization.StringDeserializer
  18. spring.kafka.consumer.value-deserializer = org.apache.kafka.common.serialization.StringDeserializer
  19. spring.kafka.consumer.auto-offset-reset = earliest
  20. spring.kafka.consumer.enable-auto-commit = true
  21. # listenner, 标识消费者监听的个数
  22. spring.kafka.listener.concurrency = 8
  23. # topic的名字
  24. kafka.topic1 = topic1

3. 生产者

  1. import lombok.extern.slf4j.Slf4j;
  2. import org.springframework.beans.factory.annotation.Value;
  3. import org.springframework.kafka.core.KafkaTemplate;
  5. @Service
  6. @Slf4j
  7. public class MyKafkaProducerServiceImpl implements MyKafkaProducerService {
  8. @Resource
  9. private KafkaTemplate<String, String> kafkaTemplate;
  10. // 读取配置文件
  11. @Value(“${kafka.topic1}”)
  12. private String topic;
  14. @Override
  15. public void sendKafka() {
  16. kafkaTemplate.send(topic, “hell world”);
  17. }
  18. }

4. 消费者

  1. @Component
  2. @Slf4j
  3. public class MyKafkaConsumer {
  4. @KafkaListener(topics = “${kafka.topic1}“)
  5. public void listen(ConsumerRecord<?, ?> record) {
  6. Optional<?> kafkaMessage = Optional.ofNullable(record.value());
  7. if (kafkaMessage.isPresent()) {
  8. log.info(“—————– record =” + record);
  9. log.info(“—————— message =” + kafkaMessage.get());