前端之家

  1. 首页
  2. 问答

Apache Camel AMQ-无法将文件写入队列-对等/客户端重置连接

• 问答

先谢谢！
我正在尝试/需要将文件从FTP写入AMQ队列。 原因-我试图使用Camel JMS AMQ在路由上添加故障转移。 我是Apache activeMQ JMS的新手。我在2个单独的节点上有2个AMQ经纪人。在其他2个节点/服务器上，我有JBoss Fuse Karaf Containers客户端应用程序。我正在从客户连接到经纪人。 AMQ控制台，日志等。但是，我无法从FTP或电子邮件路由将文件写入队列。我猜我做错了事，希望您能解决这个问题。我想知道这样做是否可以完成？

AMQ Broker节点配置片段-两者相同 我尝试删除任何相关的超时。没有差异。

activemq.xml 

    <destinationPolicy>
        <policyMap>
          <policyEntries>
            <policyEntry topic=">" producerFlowControl="true">
              <pendingMessageLimitStrategy>
                <constantPendingMessageLimitStrategy limit="1000"/>
              </pendingMessageLimitStrategy>
            </policyEntry>
            <policyEntry queue=">" producerFlowControl="true" memoryLimit="1gb">
            </policyEntry>
          </policyEntries>
        </policyMap>
    </destinationPolicy>

    <systemUsage>
        <systemUsage>
            <memoryUsage>
                <memoryUsage percentOfJvmHeap="90"/>
            </memoryUsage>
            <storeUsage>
                <storeUsage limit="100 gb"/>
            </storeUsage>
            <tempUsage>
                <tempUsage limit="50 gb"/>
            </tempUsage>
        </systemUsage>
    </systemUsage>

    <transportConnectors>
        <transportConnector name="openwire" uri="tcp://10.141.145.173:61617?connectionTimeout=0&amp;keepAlive=true&amp;useInactivityMonitor=false&amp;wireFormat.maxInactivityDuration=0&amp;enableStatusMonitor=true"/>
    </transportConnectors>

节点2 

<transportConnectors>
            <transportConnector name="openwire" uri="tcp://10.141.128.182:61617?connectionTimeout=0&amp;keepAlive=true&amp;useInactivityMonitor=false&amp;wireFormat.maxInactivityDuration=0&amp;enableStatusMonitor=true"/>
        </transportConnectors>

SYSTEM.cfg 

# node 1
activemq.port = 61617
#activemq.host = localhost
activemq.host = 10.141.145.173
activemq.url = tcp://${activemq.host}:${activemq.port}

#
# activemq configuration node 2
#
activemq.port = 61617
#activemq.host = localhost
activemq.host = 10.141.128.182
activemq.url = tcp://${activemq.host}:${activemq.port}

使用故障转移传输的客户端

jmsSourceDestination=Fleet.InboundFile.Queue
clusteredJmsMaximumRedeliveries=5
amq.url=failover://(tcp://10.141.145.173:61617,tcp://10.141.128.182:61617)?initialReconnectDelay=1000&randomize=false&timeout=5000
amq.username=admin
amq.password=admin


<transportConnectors>
    <transportConnector name="openwire" uri="tcp://10.141.145.173:61617?connectionTimeout=0&keepAlive=true&useInactivityMonitor=false&wireFormat.maxInactivityDuration=0&amp;enableStatusMonitor=true"/>
</transportConnectors>

<transportConnectors>
    <transportConnector name="openwire" uri="tcp://10.141.145.173:61617?connectionTimeout=0&keepAlive=true&useInactivityMonitor=false&wireFormat.maxInactivityDuration=0&amp;enableStatusMonitor=true"/>
</transportConnectors>

形成路线-我有一个动态路线构建器，在其中放置配置以创建路线

@Override
public void process(Exchange exchange) throws Exception {

    final String endpointConfigurationStr = exchange.getIn().getBody(String.class);
    LOG.info(endpointConfigurationStr);
    final String fileName = (String) exchange.getIn().getHeader("CamelFileName");

    if ((null != endpointConfigurationStr) && (null != fileName)) {

        Properties props = new Properties();
        props.load(new StringReader(endpointConfigurationStr));

        if (validateProperties(props)) {

            final String decoderName = props.getProperty("decoderName");
            LOG.info("DECODER NAME: " + decoderName);

            final String fileNameNoExtension = fileName.substring(0,fileName.lastIndexOf('.'));

            final String routeIdStr = String.format("fleet.inboundFile.%s.%s.Route",fileNameNoExtension,props.getProperty("transport"));

            if (props.getProperty("action").equalsIgnoreCase("activate")) {

                ServiceStatus routeStatus = exchange.getcontext().getRouteStatus(routeIdStr);

                if ((null == routeStatus) || (routeStatus.isStopped())) {
                    exchange.getcontext().addRoutes(new EndpointFileRouteBuilder(routeIdStr,EndpointDescriptorFactory(props),props.getProperty("errorArchive"),props.getProperty("unhandledArchive"),destinationEndpoint,decoderName) );
                } else {
                    LOG.info("Route " + routeIdStr + " already started");
                }
            } else if (props.getProperty("action").equalsIgnoreCase("deactivate")) {

                ServiceStatus routeStatus = exchange.getcontext().getRouteStatus(routeIdStr);
                if (routeStatus.isStarted()) {
                    exchange.getcontext().stopRoute(routeIdStr);
                } else {
                    LOG.debug("Route " + routeIdStr + " already stopped");
                }
            } else {
                LOG.error("Invalid action in File Properties");
            }
        } else {
            LOG.error("Invalid Properties File ");
        }
    } else {
        LOG.error("File Configuration File or File Name is null");
    }

}

从STR到STR的路线

    if (validateConfiguration()) {

        switch (transport.toLowerCase()) {
        case "ftp":
                    fromStr = String.format("%s://%s@%s:%s/%s?password=RAW(%s)&recursive=%s&stepwise=%s&useList=%s&passiveMode=%s&disconnect=%s"
                    + "&move=.processed"
                    + "&maxMessagesPerPoll=1"
                    + "&eagerMaxMessagesPerPoll=false"
                    + "&sortBy=file:modified"
                    + "&sendEmptyMessageWhenIdle=false"
                    + "&delay=60000"
                    + "&initialDelay=60000"
                    + "&connectTimeout=15000"
                    + "&localWorkDirectory=/tmp"
                    + "&readLockMinLength=0",transport,username,host,port,path,password,recursive,stepwise,useList,passiveMode,disconnect);
                    break;

        case "sftp":
                    fromStr = String.format("%s://%s@%s:%s/%s?password=RAW(%s)&recursive=%s&stepwise=%s&useList=%s&passiveMode=%s&disconnect=%s"
                    + "&move=.processed"
                    + "&maxMessagesPerPoll=1"
                    + "&eagerMaxMessagesPerPoll=false"
                    + "&sortBy=file:modified"
                    + "&sendEmptyMessageWhenIdle=false"
                    + "&delay=60000"
                    + "&initialDelay=60000"
                    + "&connectTimeout=15000"
                    + "&localWorkDirectory=/tmp"
                    + "&readLockMinLength=0",disconnect);
                    break;
        case "file":
                    fromStr = String.format("%s://%s/?recursive=%s"
                    + "&move=.processed"
                    + "&readLock=changed"
                    + "&maxMessagesPerPoll=1"
                    + "&sortBy=file:modified"
                    + "&delay=60000"
                    + "&initialDelay=60000"
                    + "&renameUsingCopy=true",recursive);
                    break;
        default:
            LOG.info("Unsupported transport,cannot establish from or source endpoint!");
            throw new UnsupportedTransportException("Unsupported transport,cannot establish from or source endpoint!");
        }

        // Format the To Endpoint from Parameter(s).
        final String toStr = String.format("%s",toEndpoint);
        LOG.info("*** toStr - toEndpoint: " + toStr);

路线创建

            if (Boolean.parseBoolean(isEncryptedWithCompression)) { 
                //Compression and Encryption
                PGPDataFormat pgpVerifyAndDecrypt = new PGPDataFormat();
                pgpVerifyAndDecrypt.setKeyFileName("keys/secring.gpg");
                pgpVerifyAndDecrypt.setKeyUserid(pgpKeyUserId);
                pgpVerifyAndDecrypt.setPassword(pgpPassword);
                pgpVerifyAndDecrypt.setarmored(Boolean.parseBoolean(pgpArmored));
                pgpVerifyAndDecrypt.setSignatureKeyFileName("keys/pubring.gpg");
                pgpVerifyAndDecrypt.setSignatureKeyUserid(pgpKeyUserId);
                pgpVerifyAndDecrypt.setSignatureVerificationOption(PGPKeyaccessDataFormat.SIGNATURE_VERIFICATION_OPTION_IGNORE);

                from(fromStr)
//              .routeId("Compression.with.Encryption")
                .routeId(routeId)
                .log("Message received ${file:name} for Compression and Encryption " + " from host " + host)
                .unmarshal(pgpVerifyAndDecrypt).split(new ZipSplitter())
                .streaming().convertBodyTo(String.class)
//              .wiretap("file:" + fileArchive)
                .split(body()).streaming()
                .process(new EndpointParametersProcessor(decoderName))
                .to(toStr);

            } else if (Boolean.parseBoolean(isEncryptedOnly)) { 
                //Encryption Only
                PGPDataFormat pgpVerifyAndDecrypt = new PGPDataFormat();
                pgpVerifyAndDecrypt.setKeyFileName("keys/secring.gpg");
                pgpVerifyAndDecrypt.setKeyUserid(pgpKeyUserId);
                pgpVerifyAndDecrypt.setPassword(pgpPassword);
                pgpVerifyAndDecrypt.setarmored(Boolean.parseBoolean(pgpArmored));
                pgpVerifyAndDecrypt.setSignatureKeyFileName("keys/pubring.gpg");
                pgpVerifyAndDecrypt.setSignatureKeyUserid(pgpKeyUserId);
                pgpVerifyAndDecrypt.setSignatureVerificationOption(PGPKeyaccessDataFormat.SIGNATURE_VERIFICATION_OPTION_IGNORE);

                from(fromStr)
//              .routeId("Encryption.Only")
                .routeId(routeId)
                .log("Message received ${file:name} for Encryption Only " + " from host " + host)
                .unmarshal(pgpVerifyAndDecrypt)
                .convertBodyTo(String.class)
                .choice()
                .when(simple("${header.CamelFileName} ends with 'gpg'"))
                .setHeader("CamelFileName",simple("${file:name.noext.single}"))
//              .wiretap("file:" + fileArchive)
                .split(body()).streaming()
                .process(new EndpointParametersProcessor(decoderName))
                .to(toStr);

            } else if (Boolean.parseBoolean(isCompressedOnly)) { //Only Zipped or Compressed

                ZipFileDataFormat zipFile = new ZipFileDataFormat();
                zipFile.setUsingIterator(true);

                from(fromStr)
                .routeId(routeId)
//              .routeId("Zipped.Only")
                .log(LoggingLevel.INFO,"Message received ${file:name} for Only Zipped or Compressed files from host " + host)
                .unmarshal(zipFile)
                .split(body(Iterator.class))
                .streaming()
                .convertBodyTo(String.class)
//              .wiretap("file:" + fileArchive)
                .split(body().tokenize("\n"),new FleetaggregationStrategy()).streaming()
                .process(new EndpointParametersProcessor(decoderName))
                .end()
                .choice()
                .when(simple("${body.length} > '0'" ))
                .to(toStr)
                .end();

            } else {
                //No Compression No Encryption Basic plain data file
                from(fromStr)
                .routeId(routeId)
                .log(LoggingLevel.INFO,"Message Received for No Compression No Encryption Basic plain data file " +  " from " + host)
//              .wiretap("file:" + fileArchive)
                .split(body()).streaming()
                .process(new EndpointParametersProcessor(decoderName))
                .to(toStr);
            }

TOSTR dynamic property
destinationEndpoint=activemq:queue:Fleet.InboundFile.Queue

有效路线

<route xmlns="http://camel.apache.org/schema/spring" customId="true" id="fleet.inboundFile.gcms-new-activate.sftp.Route">
    <from uri="sftp://500100471@gemft.corporate.ge.com:10022/fromvan/gary?password=RAW(+W93j2Wa)&amp;recursive=false&amp;stepwise=false&amp;useList=true&amp;passiveMode=true&amp;disconnect=false&amp;move=.processed&amp;maxMessagesPerPoll=1&amp;eagerMaxMessagesPerPoll=false&amp;sortBy=file:modified&amp;sendEmptyMessageWhenIdle=false&amp;delay=60000&amp;initialDelay=60000&amp;connectTimeout=15000&amp;localWorkDirectory=/tmp&amp;readLockMinLength=0"/>
    <onException id="onException11">
        <to id="to14" uri="file:/GlobalScapeSftpRepo/Fleet/ge-digital/fleet/core/error"/>
    </onException>
    <onException id="onException13">
        <to id="to16" uri="file:///GlobalScapeSftpRepo/Fleet/ge-digital/fleet/core/unhandled"/>
    </onException>
    <log id="log15" loggingLevel="INFO" message="Message received ${file:name} for Only Zipped or Compressed files from host gemft.corporate.ge.com"/>
    <unmarshal id="unmarshal1">
        <gzip xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:type="dataFormat"/>
    </unmarshal>
    <split id="split1" streaming="true">
        <simple>${bodyAs(java.util.Iterator)}</simple>
        <convertBodyTo id="convertBodyTo1" type="java.lang.String"/>
        <split id="split2" streaming="true">
            <expressionDefinition>tokenize(simple{${body}},)</expressionDefinition>
            <process id="process1"/>
        </split>
        <choice id="choice1">
            <when id="when1">
                <simple>${body.length} &gt; '0'</simple>
                <to id="to18" uri="activemq:queue:Fleet.InboundFile.Queue"/>
            </when>
        </choice>
    </split>
</route>

我在Broker节点上得到了什么

11:13:15,119 | WARN  | d]-nio2-thread-2 | ServerSession                    | 156 - org.apache.sshd.core - 0.14.0.redhat-001 | Exception caught
java.io.IOException: Connection reset by peer
        at sun.nio.ch.FileDispatcherImpl.read0(Native Method)[:1.7.0_241]
        at sun.nio.ch.SocketDispatcher.read(SocketDispatcher.java:39)[:1.7.0_241]
        at sun.nio.ch.IOUtil.readIntoNativeBuffer(IOUtil.java:223)[:1.7.0_241]
        at sun.nio.ch.IOUtil.read(IOUtil.java:197)[:1.7.0_241]
        at sun.nio.ch.UnixAsynchronousSocketChannelImpl.finishRead(UnixAsynchronousSocketChannelImpl.java:387)[:1.7.0_241]
        at sun.nio.ch.UnixAsynchronousSocketChannelImpl.finish(UnixAsynchronousSocketChannelImpl.java:191)[:1.7.0_241]
        at sun.nio.ch.UnixAsynchronousSocketChannelImpl.onEvent(UnixAsynchronousSocketChannelImpl.java:213)[:1.7.0_241]
        at sun.nio.ch.EPollPort$EventHandlerTask.run(EPollPort.java:293)[:1.7.0_241]
        at java.lang.Thread.run(Thread.java:748)[:1.7.0_241]

似乎客户端正在关闭连接而不传输文件？可以这样或完全做到这一点。文件大小从500K到1 GB。
尝试xfr后系统会挂起，必须回收。

w513166440 回答：Apache Camel AMQ-无法将文件写入队列-对等/客户端重置连接
暂时没有好的解决方案，如果你有好的解决方案，请发邮件至：iooj@foxmail.com
amqapache-cameljms
本文链接：https://www.f2er.com/3055672.html

大家都在问