solr自己设置hadler
solr自己设置hadler
泡海椒 发表于5个月前
solr自己设置hadler
  • 发表于 5个月前
  • 阅读 17
  • 收藏 0
  • 点赞 0
  • 评论 0

标题:腾讯云 新注册用户域名抢购1元起>>>   

package com.solr.handler;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;

import org.apache.commons.httpclient.HttpClient;
import org.apache.commons.httpclient.HttpException;
import org.apache.commons.httpclient.HttpStatus;
import org.apache.commons.httpclient.MultiThreadedHttpConnectionManager;
import org.apache.commons.httpclient.UsernamePasswordCredentials;
import org.apache.commons.httpclient.auth.AuthScope;
import org.apache.commons.httpclient.methods.GetMethod;
import org.apache.commons.httpclient.methods.PostMethod;
import org.apache.solr.common.SolrException;
import org.apache.solr.common.params.SolrParams;
import org.apache.solr.common.util.JavaBinCodec;
import org.apache.solr.common.util.NamedList;
import org.apache.solr.core.SolrCore;
import org.apache.solr.core.SolrEventListener;
import org.apache.solr.core.SolrInfoMBean;
import org.apache.solr.handler.RequestHandlerBase;
import org.apache.solr.request.SolrQueryRequest;
import org.apache.solr.response.SolrQueryResponse;
import org.apache.solr.search.SolrIndexSearcher;
import org.apache.solr.util.plugin.SolrCoreAware;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 *  解决多个solr server 共用一个索引目录时,当其中一个server commit后其他solr server需要reopen才能查询到数据
 *       <requestHandler name="/indexversion" class="com.paic.solr.handler.SharedIndexHandler" >
                <str name="notifyurl">http://localhost:7001/solr/indexversion;http://hcd-it-218:8983/solr/indexversion</str>
                <str name="way">update?commit=true</str>
         </requestHandler>  
 *
 *
 */
public class SharedIndexHandler extends RequestHandlerBase implements SolrCoreAware {
    private static final Logger LOG = LoggerFactory.getLogger(SharedIndexHandler.class.getName());
    private SolrCore core;
    private static HttpClient client;
    private HttpClient myHttpClient;
    private volatile Long indexVersion = 0L;

    private static synchronized HttpClient createHttpClient(String connTimeout, String readTimeout) {
        if (connTimeout == null && readTimeout == null && client != null)
            return client;
        MultiThreadedHttpConnectionManager mgr = new MultiThreadedHttpConnectionManager();
        // Keeping a very high number so that if you have a large number of cores
        // no requests are kept waiting for an idle connection.
        mgr.getParams().setDefaultMaxConnectionsPerHost(10000);
        mgr.getParams().setMaxTotalConnections(10000);
        mgr.getParams().setSoTimeout(readTimeout == null ? 20000 : Integer.parseInt(readTimeout)); //20 secs
        mgr.getParams().setConnectionTimeout(connTimeout == null ? 5000 : Integer.parseInt(connTimeout)); //5 secs
        HttpClient httpClient = new HttpClient(mgr);
        if (client == null && connTimeout == null && readTimeout == null)
            client = httpClient;
        return httpClient;
    }

    @Override
    public void inform(SolrCore core2) {
        this.core = core2;
        String urls = (String) initArgs.get(NOTIFYURL);
        if (urls == null || urls.trim().length() == 0)
            return;
        List<String> notifyurl = new ArrayList<String>();
        String[] urlsArray = urls.split(";");
        String hostip = null;
        try {
            hostip = java.net.InetAddress.getLocalHost().getHostAddress();
        } catch (java.net.UnknownHostException e) {
            LOG.error(e.getMessage(), e);
        }
        for (int i = 0; i < urlsArray.length; i++) {
            if (urlsArray[i].indexOf(hostip) > -1)
                continue;
            notifyurl.add(urlsArray[i]);
        }
        String way = (String) initArgs.get(NOTIFYWAY);

        String connTimeout = (String) initArgs.get(HTTP_CONN_TIMEOUT);
        String readTimeout = (String) initArgs.get(HTTP_READ_TIMEOUT);
        String httpBasicAuthUser = (String) initArgs.get(HTTP_BASIC_AUTH_USER);
        String httpBasicAuthPassword = (String) initArgs.get(HTTP_BASIC_AUTH_PASSWORD);
        myHttpClient = createHttpClient(connTimeout, readTimeout);
        if (httpBasicAuthUser != null && httpBasicAuthPassword != null) {
            myHttpClient.getState().setCredentials(AuthScope.ANY,
                    new UsernamePasswordCredentials(httpBasicAuthUser, httpBasicAuthPassword));
        }
        core.registerNewSearcherListener(getEventListener(notifyurl, way, myHttpClient));
        Map<String, SolrInfoMBean> reg = core.getInfoRegistry();
        SolrInfoMBean m = reg.get("searcher");
        try {
            indexVersion = (Long) m.getStatistics().get("indexVersion");
        } catch (Exception e) {
            LOG.error(e.getMessage(), e);
        }
    }

    @Override
    public void handleRequestBody(SolrQueryRequest req, SolrQueryResponse rsp) throws Exception {
        rsp.setHttpCaching(false);
        final SolrParams solrParams = req.getParams();
        String command = solrParams.get(COMMAND);
        if (command == null) {
            rsp.add(STATUS, OK_STATUS);
            rsp.add("message", "No command");
            return;
        } else if (command.equals(CMD_INDEX_VERSION)) {
            rsp.add(CMD_INDEX_VERSION, indexVersion);
        } else if (command.equals(CMD_SHARED_DATAIMPORT_STATUS)) {
            String urls = (String) initArgs.get(NOTIFYURL);
            String[] urlsArray = urls.split(";");
            //检查dataimport是否完成
            //http://localhost:7001/solr/indexversion
            //http://localhost:8983/solr/dataimport?command=status
            String dataimportStatus = DATA_IMPORT_IDLE_STAUTS;
            for (String url : urlsArray) {
                if(url==null||url.length()==0) continue;
                String queryUrl = url.substring(0, url.lastIndexOf("/") + 1) + "dataimport";
                dataimportStatus = queryDataimportStatus(queryUrl);
                if (!dataimportStatus.equals(DATA_IMPORT_IDLE_STAUTS))
                    break;
            }
            rsp.add("dataimporstatus", dataimportStatus);
        }
    }

    private String queryDataimportStatus(String queryUrl) {
        GetMethod getMethod = new GetMethod(queryUrl);
        getMethod.setQueryString("command=status&wt=javabin");
        String dataimportStatus = DATA_IMPORT_IDLE_STAUTS;
        try {
            int requestStatus = myHttpClient.executeMethod(getMethod);
            if (requestStatus == HttpStatus.SC_OK) {
                NamedList nl = (NamedList) new JavaBinCodec().unmarshal(getMethod.getResponseBodyAsStream());
                dataimportStatus = (String) nl.get("status");
                System.out.println(dataimportStatus);
            }
        } catch (HttpException e) {
            LOG.error("查询dataimport状态错误。", e);
        } catch (IOException e) {
            LOG.error("查询dataimport状态错误。", e);
        } finally {
            getMethod.releaseConnection();
        }
        return dataimportStatus;
    }

    @Override
    public String getDescription() {
        return "SharedIndexHandler used by multiple Solr servers and a shared index. to notify each other";
    }

    @Override
    public String getSourceId() {
        return "$Id: SharedIndexHandler.java 2012-09-15 $";
    }

    @Override
    public String getSource() {
        return "$URL: http://svn.apache.org/repos/asf/lucene/dev/tags/lucene_solr_3_6_0/com/paic/solr/handler/handler/SharedIndexHandler.java $";
    }

    @Override
    public String getVersion() {
        return "$Revision: 001 $";
    }

    /**
     *  在post commit执行时先判断远程服务器的版本和本地是否一致,
     *  不一致需要触发commit请求
     * @param notifyurls
     * @return
     */
    private SolrEventListener getEventListener(final List<String> notifyurls, final String way,
            final HttpClient httpclient) {
        return new SolrEventListener() {
            @Override
            public void init(NamedList args) {

            }

            @Override
            public void postCommit() {

            }

            @Override
            public void newSearcher(SolrIndexSearcher newSearcher, SolrIndexSearcher currentSearcher) {
                indexVersion = (Long) newSearcher.getStatistics().get("indexVersion");
                Thread t = new Thread(new Runnable() {
                    @Override
                    public void run() {
                        for (String url : notifyurls) {
                            NamedList response = null;
                            try {
                                response = getLatestVersion(url);
                            } catch (Exception e) {
                                LOG.error(" 获取索引版本信息失败: " + e.getMessage());
                                continue;
                            }
                            long latestVersion = (Long) response.get(CMD_INDEX_VERSION);
                            LOG.info(url + ":" + latestVersion + "   本地:" + indexVersion);
                            if (indexVersion == latestVersion) {
                                continue;
                            }
                            String notifyurl = url.substring(0, url.lastIndexOf("/") + 1) + way;
                            LOG.info("need to notify=" + notifyurl);
                            GetMethod getMethod = new GetMethod(notifyurl);
                            try {
                                int statusCode = httpclient.executeMethod(getMethod);
                                if (statusCode == HttpStatus.SC_OK) {
                                    LOG.info("请求:" + notifyurl + "成功.");
                                }
                            } catch (HttpException e) {
                                LOG.error("触发请求" + notifyurl + "失败");
                            } catch (IOException e) {
                                LOG.error("触发请求" + notifyurl + "失败");
                            } finally {
                                getMethod.releaseConnection();
                            }
                        }
                    }
                });
                t.start();

            }

            @SuppressWarnings("unchecked")
            NamedList getLatestVersion(String notifyUrl) throws IOException {
                PostMethod post = new PostMethod(notifyUrl);
                post.addParameter(COMMAND, CMD_INDEX_VERSION);
                post.addParameter("wt", "javabin");
                return getNamedListResponse(post);
            }

            private NamedList getNamedListResponse(PostMethod method) throws IOException {
                try {
                    int status = httpclient.executeMethod(method);
                    if (status != HttpStatus.SC_OK) {
                        throw new SolrException(SolrException.ErrorCode.SERVICE_UNAVAILABLE,
                                "Request failed for the url " + method);
                    }
                    return (NamedList) new JavaBinCodec().unmarshal(method.getResponseBodyAsStream());
                } finally {
                    try {
                        method.releaseConnection();
                    } catch (Exception e) {
                    }
                }
            }
        };
    }

    public static final String NOTIFYURL = "notifyurl";
    public static final String NOTIFYWAY = "way";
    public static final String HTTP_CONN_TIMEOUT = "httpConnTimeout";
    public static final String COMMAND = "command";
    public static final String HTTP_READ_TIMEOUT = "httpReadTimeout";
    public static final String HTTP_BASIC_AUTH_USER = "httpBasicAuthUser";
    public static final String HTTP_BASIC_AUTH_PASSWORD = "httpBasicAuthPassword";
    public static final String STATUS = "status";
    public static final String ERR_STATUS = "ERROR";
    public static final String OK_STATUS = "OK";
    public static final String CMD_INDEX_VERSION = "indexversion";
    public static final String CMD_SHARED_DATAIMPORT_STATUS = "dataimportstatus";
    public static final String DATA_IMPORT_IDLE_STAUTS = "idle";

}
 

共有 人打赏支持
粉丝 10
博文 254
码字总数 289006
×
泡海椒
如果觉得我的文章对您有用,请随意打赏。您的支持将鼓励我继续创作!
* 金额(元)
¥1 ¥5 ¥10 ¥20 其他金额
打赏人
留言
* 支付类型
微信扫码支付
打赏金额:
已支付成功
打赏金额: