查看: 68688|回复: 23451
打印 上一主题 下一主题

所以将会参考Netty中的有界无锁队列

[复制链接]
致富网赚联盟
跳转到指定楼层
楼主
发表于 2018-10-17 00:25:31 | 只看该作者 |只看大图 回帖奖励 |倒序浏览 |阅读模式

除了文殊普贤二人,其余佛门中人也是大惊失色,这法术乃是如来秘授给亲信之人的,只道威力无比,非到万不得已不可使出来,不想却只是样子吓人,实则毫无作用。

怎样可以赚到美元

“实在是太过分了。”拉克丝目光骤然一缩,眼神变得冰冷无比,这一种完全将士兵,将人当成棋子牺牲掉的手段让她十分的愤怒,虽然双方是敌人,上了战场不是你死就是我亡,但是这一种手段却还是深深的刺激到了拉克丝,在这些光束射过来的瞬间拉克丝忍不住高声娇喝道。
李小婉压低声音,道:“方家有可能是隐藏在世间的修真家族。我们特勤局目前只记录了华国十大古武世家的信息,另外还有一些世外高人,但修真一系,却是没有记录。所以对于这个方家,我们也是知之甚少。”

哪里去了?悟空第一个便想到如来,镇元子是如来选出来看护人参果树的,如果镇元子有何异动,除了如来还会是旁人所为吗?

所以将会参考Netty中的有界无锁队列


在互联网设计架构过程中,日志异步落库,俨然已经是高并发环节中不可缺少的一环。为什么说是高并发环节中不可缺少的呢? 原因在于,如果直接用mq进行日志落库的时候,低并发下,生产端生产数据,然后由消费端异步落库,是没有什么问题的,而且性能也都是异常的好,估计tp99应该都在1ms以内。但是一旦并发增长起来,慢慢的你就发现生产端的tp99一直在增长,从1ms,变为2ms,4ms,直至send timeout。尤其在大促的时候,我司的系统就经历过这个情况,当时mq的发送耗时超过200ms,甚至一度有不少timeout产生。

考虑到这种情况在高并发的情况下才出现,所以今天我们就来探索更加可靠的方法来进行异步日志落库,保证所使用的方式不会因为过高的并发而出现接口ops持续下降甚至到不可用的情况。

 

方案一: 基于log4j的异步appender实现

此种方案,依赖于log4j。在log4j的异步appender中,通过mq进行生产消费入库。相当于在接口和mq之间建立了一个缓冲区,使得接口和mq的依赖分离,从而不让mq的操作影响接口的ops。

此种方案由于使用了异步方式,且由于异步的discard policy策略,当大量数据过来,缓冲区满了之后,会抛弃部分数据。此种方案适用于能够容忍数据丢失的业务场景,不适用于对数据完整有严格要求的业务场景。

来看看具体的实现方式:

首先,我们需要自定义一个Appender,继承自log4j的AppenderSkeleton类,实现方式如下:

public class AsyncJmqAppender extends AppenderSkeleton {

    @Resource(name = "messageProducer")
    private MessageProducer messageProducer;

    @Override
    protected void append(LoggingEvent loggingEvent) {
        asyncPushMessage(loggingEvent.getMessage());
    }

    /**
     * 异步调用jmq输出日志
     * @param message
     */
    private void asyncPushMessage(Object message) {

        CompletableFuture.runAsync(() -> {

            Message messageConverted = (Message) message;

            try {
                messageProducer.send(messageConverted);
            } catch (JMQException e) {
                e.printStackTrace();
            }

        });
    }


    @Override
    public boolean requiresLayout() {
        return false;
    }

    @Override
    public void close() {

    }
}

然后在log4j.xml中,为此类进行配置:

<!--异步JMQ appender-->
<appender name="async_mq_appender" class="com.jd.limitbuy.common.util.AsyncJmqAppender">
    <!-- 设置File参数:日志输出文件名 -->
    <param name="File" value="D:/export/Instances/order/server1/logs/order.async.jmq" />
    <!-- 设置是否在重新启动服务时,在原有日志的基础添加新日志 -->
    <param name="Append" value="true" />
    <!-- 设置文件大小 -->
    <param name="MaxFileSize" value="10KB" />
    <!-- 设置文件备份 -->
    <param name="MaxBackupIndex" value="10000" />
    <!-- 设置输出文件项目和格式 -->
    <layout class="org.apache.log4j.PatternLayout">
        <param name="ConversionPattern" value="%m%n" />
    </layout>
</appender>
<logger name="async_mq_appender_logger">
    <appender-ref ref="async_mq_appender"/>
</logger>

最后就可以按照如下的方式进行正常使用了:

private static Logger logger = LoggerFactory.getLogger("filelog_appender_logger");

注意: 此处需要注意log4j的一个性能问题。在log4j的conversionPattern中,匹配符最好不要出现 C% L%通配符,压测实践表明,这两个通配符会导致log4j打日志的效率降低10倍。

方案一很简便,且剥离了接口直接依赖mq导致的性能问题。但是无法解决数据丢失的问题(但是我们其实可以在本地搞个策略落盘来不及处理的数据,可以大大的减少数据丢失的几率)。但是很多的业务场景,是需要数据不丢失的,所以这就衍生出我们的另一套方案来。

 

方案二:增量消费log4j日志

此种方式,是开启worker在后台增量消费log4j的日志信息,和接口完全脱离。此种方式相比方案一,可以保证数据的不丢失,且可以做到完全不影响接口的ops。但是此种方式,由于是后台worker在后台启动进行扫描,会导致落库的数据慢一些,比如一分钟之后才落库完毕。所以适用于对落库数据实时性不高的场景。

具体的实现步骤如下:

首先,将需要进行增量消费的日志统一打到一个文件夹,以天为单位,每天生成一个带时间戳日志文件。由于log4j不支持直接带时间戳的日志文件生成,所以这里需要引入log4j.extras组件,然后配置log4j.xml如下:

image

之后在代码中的申明方式如下:

private static Logger businessLogger = LoggerFactory.getLogger("file_rolling_logger");

最后在需要记录日志的地方使用方式如下:

businessLogger.error(JsonUtils.toJSONString(myMessage))

这样就可以将日志打印到一个单独的文件中,且按照日期,每天生成一个。

然后,当日志文件生成完毕后,我们就可以开启我们的worker进行增量消费了,这里的增量消费方式,我们选择RandomAccessFile这个类来进行,由于其独特的位点读取方式,可以使得我们非常方便的根据位点的位置来消费增量文件,从而避免了逐行读取这种低效率的实现方式。

注意,为每个日志文件都单独创建了一个位点文件,里面存储了对应的文件的位点读取信息。当worker扫描开始的时候,会首先读取位点文件里面的位点信息,然后找到相应的日志文件,从位点信息位置开始进行消费。这就是整个增量消费worker的核心。具体代码实现如下(代码太长,做了折叠):

/**
 * @Description: 增量日志扫描worker
 * @Detail: 此worker主要用来扫描增量日志,日志本身会在不停的插入中,此worker会不停的扫描此日志来将数据上传到kafka集群
 * @date 2018-04-08 10:30
 */
public class LimitBuyScanWorker {

    /**
     * 日志和位点文件保存的目录
     */
    private static final String FILE_DIRECTORY = "D:\export\Instances\order\server1\logs\";

    /**
     * 每次步进的长度,此处为1000行
     */
    private static final int SCAN_STEP = 1000;

    /**
     * 日志文件名前缀
     */
    private static final String LOG_FILE_PREFIX = "limitbuy.soa.order.";

    /**
     * 位点文件名后缀
     */
    private static final String OFT_FILE_APPENDIX = ".offset";

    public void logScanner() {

        //当前时间
        Date currentDate = new Date();

        //今日
        String currentDay = DateUtil.formatDate("yyyy-MM-dd", currentDate);
        //今日日志文件路径
        String currentLogFilePath = FILE_DIRECTORY + LOG_FILE_PREFIX + currentDay;
        logger.error("今日的日志文件路径:" + currentLogFilePath);
        //今日位点文件路径
        String currentOffsetFilePath = FILE_DIRECTORY + LOG_FILE_PREFIX + currentDay + OFT_FILE_APPENDIX;

        //昨日
        String yesterDay = DateUtil.formatDate("yyyy-MM-dd", DateUtil.queryPlusDay(currentDate, -1));
        //昨日日志文件路径
        String yesterdayLogFilePath = FILE_DIRECTORY + LOG_FILE_PREFIX + yesterDay;
        logger.error("昨日的日志文件路径:" + yesterdayLogFilePath);
        //昨日位点文件路径
        String yesterdayOffsetFilePath = FILE_DIRECTORY + LOG_FILE_PREFIX + yesterDay + OFT_FILE_APPENDIX;

        //先检测昨日位点和文件体积是否一致,不一致则代表未消费完毕
        boolean yesterdayConsumedOK = checkIfConsumeOK(yesterdayLogFilePath, yesterdayOffsetFilePath);
        logger.error("昨日的日志文件已被消费完毕:" + yesterdayConsumedOK);

        //昨日的文件已扫描完毕
        if (yesterdayConsumedOK) {
            //扫描并消费今日增量日志
            scanAndConsumeLog(currentLogFilePath, currentOffsetFilePath);
        }
        //昨日的文件未扫描完毕
        else {
            //扫描并消费昨日增量日志
            scanAndConsumeLog(yesterdayLogFilePath, yesterdayOffsetFilePath);
        }
    }

    /**
     * 检测日志是否被扫描消费完毕,true:消费完毕;false:未消费完毕
     * @Description 此举主要防止log4j在零点大促开始的时候,突然的滚动文件造成的部分增量日志不会被消费的问题
     * @param logFilePath
     * @param offsetFilePath
     */
    private boolean checkIfConsumeOK(String logFilePath, String offsetFilePath) {
        try {

            //打开文件
            RandomAccessFile randomAccessFile = new RandomAccessFile(logFilePath, "r");

            //得到当前位点
            long currentOffset = checkOffset(offsetFilePath);

            //得到文件总长
            long currentFileLength = randomAccessFile.length();

            //比对
            if (currentOffset >= currentFileLength) {
                return true;
            }
            return false;
        } catch (FileNotFoundException e) {
            logger.error("com.jd.limitbuy.service.worker.logScanner 出错(FileNotFoundException):", e);
            AlarmUtil.alarm("com.jd.limitbuy.service.worker.logScanner 出错:" + e.getMessage());
            return false;
        } catch (IOException e) {
            logger.error("com.jd.limitbuy.service.worker.logScanner 出错(IOException):", e);
            AlarmUtil.alarm("com.jd.limitbuy.service.worker.logScanner 出错:" + e.getMessage());
            return false;
        }
    }

    /**
     * 扫描并消费增量日志
     * @param logFilePath
     * @param offsetFilePath
     */
    private void scanAndConsumeLog(String logFilePath, String offsetFilePath) {
        try {

            RandomAccessFile randomAccessFile = new RandomAccessFile(logFilePath, "r");

            //得到当前位点
            long currentOffset = checkOffset(offsetFilePath);
            logger.error("开始位点==>" + currentOffset);

            //重置位点到当前位点
            if (currentOffset <= randomAccessFile.length()) {
                randomAccessFile.seek(currentOffset);
            }

            //读取@SCAN_STEP行
            for (long i = currentOffset; i < currentOffset + SCAN_STEP; i++) {
                //得到行
                String result = randomAccessFile.readLine();
                //如果内容不为空
                if (StringUtil.isNotBlank(result)) {
                    //TODO 逻辑实现
                }
            }

            //读取@SCAN_STEP行之后的位点
            logger.error("读取" + SCAN_STEP + "行之后位点==>" + randomAccessFile.getFilePointer());

            //如果update不成功,可以不处理,后面扫描进来重新过一遍即可
            updateOffset(randomAccessFile.getFilePointer(), offsetFilePath);

            logger.error("文件总长==>" + randomAccessFile.length());

        } catch (FileNotFoundException e) {
            logger.error("com.jd.limitbuy.service.worker.logScanner 出错(FileNotFoundException):", e);
            AlarmUtil.alarm("com.jd.limitbuy.service.worker.logScanner 出错:" + e.getMessage());
        } catch (IOException e) {
            logger.error("com.jd.limitbuy.service.worker.logScanner 出错(IOException):", e);
            AlarmUtil.alarm("com.jd.limitbuy.service.worker.logScanner 出错:" + e.getMessage());
        }
    }

    /**
     * 校验位点
     *     不存在则创建并赋值为0
     *     已存在则更新位点
     * @param offsetFilePath
     * @return
     * @throws IOException
     */
    private long checkOffset(String offsetFilePath) throws IOException {

        File offsetFile = new File(offsetFilePath);

        //如果位点文件不存在,则创建位点文件并返回0
        if (!offsetFile.exists()) {
            updateOffset(0, offsetFilePath);
            return 0;
        }
        //如果位点文件存在,则返回位点文件内容
        else {
            FileReader fileReader = new FileReader(offsetFilePath);
            StringBuilder stringBuilder = new StringBuilder();
            char[] bytesChar = new char[50];
            fileReader.read(bytesChar);
            fileReader.close();

            for (char c : bytesChar) {
                stringBuilder.append(c);
            }

            String filteredOffset = stringBuilder.toString().trim();

            if (StringUtil.isNotBlank(filteredOffset)) {
                return Long.parseLong(filteredOffset);
            } else {
                return 0;
            }
        }
    }

    /**
     * 更新位点信息
     * @param offset
     * @param offsetFilePath
     */
    private void updateOffset(long offset, String offsetFilePath) throws IOException {
        FileWriter fileWriter = new FileWriter(offsetFilePath);
        fileWriter.write(offset + "");
        fileWriter.flush();
        fileWriter.close();
    }
}

此种方式由于worker扫描是每隔一段时间启动一次进行消费,所以导致数据从产生到入库,可能经历时间超过一分钟以上,但是在一些对数据延迟要求比较高的业务场景,比如库存扣减,是不能容忍的,所以这里我们就引申出第三种做法,基于内存文件队列的异步日志消费。

 

方案三:基于内存文件队列的异步日志消费

由于方案一和方案二都严重依赖log4j,且方案本身都存在着要么丢数据,要么入库时间长的缺点,所以都并不是那么尽如人意。但是本方案的做法,既解决了数据丢失的问题,又解决了数据入库时间被拉长的尴尬,所以是终极解决之道。而且在大促销过程中,此种方式经历了实战检验,可以大面积的推广使用。

此方案中提到的内存文件队列,是我司自研的一款基于RandomAccessFile和MappedByteBuffer实现的内存文件队列。队列核心使用了ArrayBlockingQueue,并提供了produce方法,进行数据入管道操作,提供了consume方法,进行数据出管道操作。而且后台有一个worker一直启动着,每隔5ms或则是遍历了100条数据之后,就将数据落盘一次,以防数据丢失。具体的设计,就这么多,感兴趣的可以根据我提供的信息,自己实践一下。

由于有此中间件的加持,数据生产的时候,只需要入压入管道,然后消费端进行消费即可。未被消费的数据,会进行落盘操作,谨防数据丢失。当大促的时候,大量数据涌来的时候,管道满了的情况下会阻塞接口,数据不会被抛弃。虽然可能会导致接口在那一瞬间无响应,但是由于有落盘操作和消费操作(此操作操控的是JVM堆外内存数据,不受GC的影响,所以不会出现操作暂停的情况,为什么呢?因为用了MappedByteBuffer),此种阻塞并未影响到接口整体的ops。

在实际使用的时候,ArrayBlockingQueue作为核心队列,显然是全局加锁的,后续我们考虑升级为无锁队列,所以将会参考Netty中的有界无锁队列:MpscArrayQueue。预计性能将会再好一些。

受限于公司政策,我仅提供大致思路,但是不会提供具体代码,有问题评论区交流吧。

 

上面就是在进行异步日志消费的时候,我所经历的三个阶段,并且一步一步的优化到目前的方式。虽然过程曲折,但是结果令人欢欣鼓舞。如果喜欢就给个推荐,后续我将会持续更新你所不知道的系列,以期达到抛砖引玉的效果。

当前文章:http://www.fstdn.com/4j06n/

发布时间:2018-10-17 17:51:15

跑步就能赚钱的app 发视频网赚项目 2018年赚钱的游戏 在家玩手机赚钱开公司 网络写手挣钱 如何利用国外网站赚钱 钱多多辱华是真的吗 有没有在家就能挣钱的工作

编辑:侯纯

点击获取礼包
大话2商人月赚上万
沙发
发表于 2018-10-17 13:50:45 | 只看该作者
“你们的对手是我。”唐昊洪亮而又充满无尽霸气的声音爆响,刚才那一片碎肉正是他面对的三名封号斗罗中的一个。唐昊的动很简单,炸环。他炸开了自己的第一个魂环,瞬间爆出的攻击力根本就不是那名刚刚达到封号斗罗级别不久,只有九十一级的强攻系封号斗罗所能抵挡的,顿时被昊天锤恐怖的威力轰杀成了碎肉。
回复 支持 反对

使用道具 举报

可以赚人民币的游戏
板凳
发表于 2018-10-17 15:56:34 | 只看该作者
方潜嘿嘿一笑,走到老苏旁边,给他点上烟,道:“老苏,我发现你这人挺贼啊。看着什么都不关心,其实心里都有。”http://www.fstdn.com/news/2018101728317/index.html
回复 支持 反对

使用道具 举报

空闲时间怎么赚钱
地板
发表于 2018-10-17 15:48:14 | 只看该作者
“别问我,去问你的姑婆,你还是先回去吧。”刘皓投有直接回答马小玲反而对祝国华说况国华闻言,点了点头离开了,他来这里最主要的就是看看自己的孙子,现在都死了,他也投什么好留恋的,既然这件事己经被人知道了,还是先离开吧,马小玲想都阻止却被刘皓拦住了。
回复 支持 反对

使用道具 举报

可靠的网上兼职
5#
发表于 2018-10-17 00:20:47 | 只看该作者
一堆堆还没有被烧烬的帐篷被白雪覆盖着,(露)出黑漆漆的一角,黑丝烂布纠缠着已经毁坏的盔甲,被烧榕了一半的各种兵器横七竖八地堆在地上,一堆堆没有了箭杆的箭头,这些都还可以回炉重新打制,但让人触目惊心的是一滩滩血迹,被冻得硬硬邦邦,凝结成一块块红色的血冰,偶然还从血冰中(露)出一截被砍掉的胳膊。
回复 支持 反对

使用道具 举报

网络兼职招聘网
6#
发表于 2018-10-17 05:01:28 | 只看该作者
四大平台,只有微歌用户要么进不去,要么进去买不了,要么买了听不了。群里,微歌用户们谩骂、哀嚎声一片。
回复 支持 反对

使用道具 举报

手机暴利灰色项目600
7#
发表于 2018-10-17 08:25:53 | 只看该作者
不过不管如何,这毕竟只是角都的猜想,不过如果刘皓在的话绝对会写一个服字给角都,真不愧是老江湖,他说的事情里面起码有一半是对的。
回复 支持 反对

使用道具 举报

社会上挣钱的路子
8#
发表于 2018-10-17 03:54:32 | 只看该作者
李隆基笑了,李庆安的表态让他心中轻轻松了一口气,只要李庆安接收这一点,其他什么都好商量,他点点头笑道:“这次陇右之战有功有罪,有功者,朕会给予重赏,有罪者,朕也会给予重罚,朕一定会赏罚分明,虽然在此战中,给你定的拦截功劳为第三,但你还有其他的功劳,比如火药,就凭这一点,朕就可以在战役以外对你进行额外的封赏,你明白吗?”
回复 支持 反对

使用道具 举报

网上赚钱的几种方法
9#
发表于 2018-10-17 16:34:36 | 只看该作者
唐三听到马红俊的话,向他比了比拳头,“胖子,这你就不懂了吧。天才和疯子,从来都是一线之隔的。”
回复 支持 反对

使用道具 举报

您需要登录后才可以回帖 登录 | 立即注册

本版积分规则

犯法快速来钱的方法是互联网最大的搜索引擎优化研究中心,是致力于培养学员用户体验意识和提供专业技术解答的专业培训机构, 成立于2007年,2008年第一家入驻歪歪的培训机构,2014年成为腾讯课堂战略合作机构。
© 2007-2016 犯法快速来钱的方法 湘ICP备13004652号-1 Powered by Discuz!X  Template by 犯法快速来钱的方法 
快速回复 返回顶部 返回列表