专栏首页用户画像微博订阅评论

微博订阅评论

参考:http://open.weibo.com/wiki/%E7%A4%BA%E4%BE%8B%E4%BB%A3%E7%A0%81

微博开放平台给出的代码有很多bug,在此纠正

如乱码问题的解决: String message = new String(bytes,"UTF-8");

package com.thinkgem.jeesite.modules.util.weibo.request;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.thinkgem.jeesite.modules.util.weibo.post.WeiBoPostService;
import org.apache.commons.httpclient.HostConfiguration;
import org.apache.commons.httpclient.HttpClient;
import org.apache.commons.httpclient.HttpStatus;
import org.apache.commons.httpclient.MultiThreadedHttpConnectionManager;
import org.apache.commons.httpclient.methods.GetMethod;
import org.apache.commons.lang3.StringUtils;

import javax.servlet.ServletContextEvent;
import javax.servlet.ServletContextListener;
import java.io.ByteArrayOutputStream;
import java.io.DataInputStream;
import java.io.IOException;
import java.io.UnsupportedEncodingException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
public class ReceiveMessageService implements ServletContextListener {
    private MultiThreadedHttpConnectionManager httpConnManager;
    private HttpClient httpClient;
    private List<String> streamingUrlList=new ArrayList<String>();
    private int curStreamUrlIndex;
    public static long lastMsgLocation = -1L;
    private DataInputStream inputStream;
    private final int recBufSize = 256;
    private byte[] recBuf;
    private int recIndex;

    @Override
    public void contextInitialized(ServletContextEvent sce) {
        new ReceiveMessageService().init();
    }

    @Override
    public void contextDestroyed(ServletContextEvent sce) {

    }

    /**
     * 初始化httpclient,并启动获取数据线程
     */
    public void init() {
        this.streamingUrlList.add("http://c.api.weibo.com/datapush/comment?subid=10823");
        //this.streamingUrlList.add("http://c.api.weibo.com/datapush/status?subid=10718");
        this.httpConnManager = new MultiThreadedHttpConnectionManager();
        this.httpConnManager.getParams().setMaxConnectionsPerHost(HostConfiguration.ANY_HOST_CONFIGURATION, 10);
        this.httpConnManager.getParams().setMaxTotalConnections(10);
        this.httpConnManager.getParams().setSoTimeout(2147483647);
        this.httpConnManager.getParams().setConnectionTimeout(10000);
        this.httpConnManager.getParams().setReceiveBufferSize(655350);
        this.httpClient = new HttpClient(this.httpConnManager);
        new ReadTask().start();
    }



    /**
     * 获取数据线程
     *
     */
    class ReadTask extends Thread {

        /**
         * 启一个线程从服务器读取数据
         */
        public void run() {
            while (true) {
                GetMethod method = null;
                recIndex = 0;
                recBuf = new byte[recBufSize];
                try {
                    method = connectStreamServer();//建立连接
                    while (true) {
                        processLine();//读取并处理数据
                    }
                } catch (Exception e) {
                    // 当连接断开时,重新连接
                    lastMsgLocation = lastMsgLocation + 2;
                    curStreamUrlIndex = ++curStreamUrlIndex % streamingUrlList.size();
                } finally {
                    if (method != null)
                        method.releaseConnection();
                }
            }
        }

        /**
         * 建立http连接
         *
         * @return
         */
        private GetMethod connectStreamServer() throws IOException {
            String targetURL = (String) streamingUrlList.get(curStreamUrlIndex);
            // 从指定的since_id开始读取数据,保证读取数据的连续性,消息完整性
            if (lastMsgLocation > 0L) {
                targetURL = targetURL + "&since_id=" + lastMsgLocation;
            }
            //System.out.println("StreamingReceiver http get url=" + targetURL);
            GetMethod method = new GetMethod(targetURL);
            int statusCode = 0;
            statusCode = httpClient.executeMethod(method);

            if (statusCode != HttpStatus.SC_OK) {
                throw new RuntimeException(".. ");
            }

            try {
                inputStream = new DataInputStream(method.getResponseBodyAsStream());
            } catch (IOException e) {
                throw new RuntimeException("get stream input io exception", e);
            }

            return method;
        }

        /**
         * 读取并处理数据
         *
         * @throws IOException
         */
        private void processLine() throws IOException {
            byte[] bytes = readLineBytes();//读取数据
            if ((bytes != null) && (bytes.length > 0)) {
                String message = new String(bytes,"UTF-8");//编码格式设置
                handleMessage(message);
            }
        }

        /**
         * 可以重写此方法解析message,同时需将lastMsgLocation赋值为最近1条消息的id
         *
         * @param message
         */
        private void handleMessage(String message) throws UnsupportedEncodingException {
            JSONObject rs=JSON.parseObject(message);
            if(StringUtils.isBlank(rs.getString("errorCode"))){
                lastMsgLocation= rs.getLong("id");
                WeiBoPostService service=new WeiBoPostService();
                service.receiveMessage(message);
            }
        }

        /**
         * 读取数据
         *
         * @return
         * @throws IOException
         */
        public byte[] readLineBytes() throws IOException {
            byte[] result = null;
            ByteArrayOutputStream bos = new ByteArrayOutputStream();
            int readCount = 0;
            if ((recIndex > 0) && (read(bos))) {
                return bos.toByteArray();
            }
            while ((readCount = inputStream.read(recBuf, recIndex, recBuf.length - recIndex)) > 0) {
                recIndex = (recIndex + readCount);
                if (read(bos)) {
                    break;
                }
            }
            result = bos.toByteArray();
            if ((result == null) || ((result != null) && (result.length <= 0) && (recIndex <= 0))) {
                throw new IOException("++++ Stream appears to be dead, so closing it down");
            }
            return result;
        }

        /**
         * 读数据到bos
         *
         * @param bos
         * @return
         */
        private boolean read(ByteArrayOutputStream bos) {
            boolean result = false;
            int index = -1;
            for (int i = 0; i < recIndex - 1; i++) {
                // 13cr-回车 10lf-换行
                if ((recBuf[i] == 13) && (recBuf[(i + 1)] == 10)) {
                    index = i;
                    break;
                }
            }
            if (index >= 0) {
                bos.write(recBuf, 0, index);
                byte[] newBuf = new byte[recBufSize];
                if (recIndex > index + 2) {
                    System.arraycopy(recBuf, index + 2, newBuf, 0, recIndex - index - 2);
                }
                recBuf = newBuf;
                recIndex = (recIndex - index - 2);
                result = true;
            } else if (recBuf[(recIndex - 1)] == 13) {
                bos.write(recBuf, 0, recIndex - 1);
                Arrays.fill(recBuf, (byte) 0);
                recBuf[0] = 13;
                recIndex = 1;
            } else {
                bos.write(recBuf, 0, recIndex);
                Arrays.fill(recBuf, (byte) 0);
                recIndex = 0;
            }
            return result;
        }
    }

}

本文参与腾讯云自媒体分享计划,欢迎正在阅读的你也加入,一起分享。

我来说两句

0 条评论
登录 后参与评论

相关文章

  • 路由器题目

    1. 某公司网络拓扑如下图所示,路由器R1通过接口E1、E2分别连接局域网1、局域网2, 通过接口L0连接路由器R2,并通过路由器R2连接域名服务器与互联网。R...

    week
  • Python爬虫七麦APP榜单

    week
  • windows搭建hadoop分布式系统架构

    1、下载hadoop的安装包:http://hadoop.apache.org/->Releases->mirror site

    week
  • 技术分享 | Online DDL 工具 gh-ost

    爱可生 DBA 团队成员,负责公司 DMP 产品的运维和客户 MySQL 问题的处理。擅长数据库故障处理。对数据库技术和 python 有着浓厚的兴趣。

    爱可生开源社区
  • Tesseract ocr文字识别

    版权声明:本文为博主原创文章,未经博主允许不得转载。 https://blog.csdn.net/haluoluo211/article/details...

    bear_fish
  • 自定义工作流设计

    工作偶尔会遇到需要审批相关的系统,对于流程步骤相对固定的,一般会采取某些第三方的工作流来做对应的系统。目前唯一用过的就是activiti工作流。对它进行了简单的...

    用户5166330
  • Pytorch-ResNet(残差网络)-下

    在左图(准确率)的比较中,从AlexNet到GoogleNet再到ResNet,准确率逐渐提高。20层结构是很多网络结构性能提升的分水岭,在20层之前,模型性能...

    用户6719124
  • Go标准编译器所做的一些优化

    本博的另两篇文章中已经介绍了Go标准编译器对字符串和字节切片之间相互转换所做的一些优化和边界消除优化。下面将介绍Go标准编译器(截至Go SDK 1.12.x)...

    刘老貘
  • WebService基于SoapHeader实现安全认证

          本文仅提供通过设置SoapHeader来控制非法用户对WebService的调用,如果是WebService建议使用WSE3.0来保护Web服务,如...

    跟着阿笨一起玩NET
  • 用vc++做滚动条控件

      首先用应用向导产生一名为scro的基于对话框的应用,再利用对话框编辑器加入两个一样的水平滚动条,两个滚动条的id分别取idc—scr1和idc—scr2,结...

    用户1154259

扫码关注云+社区

领取腾讯云代金券